"""Renovation rules and frozen search orchestration; no platform/model traffic.""" from __future__ import annotations import copy import sqlite3 import pytest from fastapi import HTTPException from fastapi.testclient import TestClient from apps.server.db import Store, StoreConflict from apps.server.discovery import DiscoveryService, select_results from apps.server.learning_checks import output_assertion_checks from apps.server.tests.test_learning import client, initialize, task from apps.server.tests.test_postgres import pg_client FIELDS = [{"name": "title", "selector": ".title", "source": "text"}, {"name": "author", "selector": ".author", "source": "text"}, {"name": "url", "selector": "a", "source": "href", "required": True}] EXTRACT = {"kind": "extract", "selector": '[data-testid="result"]', "output_name": "search_results", "fields": FIELDS, "limit": 20} BASE = "http://127.0.0.1:8810" def output(items): return {"items": items, "count": len(items), "truncated": True, "source_url": BASE + "/search", "coverage": "bounded" if items else "empty"} def item(title, letter="a", *, author="作者", url=None): return {"title": title, "author": author, "url": url or BASE + "/explore/" + letter * 24} def search_candidate(client, items): created = task(client) prefix = "/api/learning/tasks/" + created["id"] worker = client.worker_reference old_capture, old_call = worker.capture, worker.call def capture(task_id): value = old_capture(task_id) for element in value["elements"]: if element["selector"] == '[data-testid="search-input"]': element["search_input"] = True return value worker.capture = capture async def call(method, params): result = await old_call(method, params) if method == "action" and params["kind"] == "extract": result["result"] = output(items) if method == "verify": result["outputs"]["search_results"] = output(items) return result worker.call = call client.post(prefix + "/browser/open") client.post(prefix + "/browser/ready", json={"identity_label": "测试账号"}) client.post(prefix + "/browser/capture", json={}) for action in ({"kind": "fill", "selector": '[data-testid="search-input"]', "parameter": "keyword"}, {"kind": "press", "selector": '[data-testid="search-input"]', "key": "Enter"}, {"kind": "wait", "selector": '[data-testid="result"]', "state": "visible", "min_count": 1}, EXTRACT): result = client.post(prefix + "/browser/action", json=action) assert result.status_code == 200, result.text candidate = client.post(prefix + "/candidates", json={}).json() assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed" return created, candidate @pytest.fixture(params=["client", "pg_client"]) def discovery_client(request): return request.getfixturevalue(request.param) def test_discovery_persists_frozen_provenance_owner_rules_and_conditions(discovery_client): client = discovery_client initialize(client) created, candidate = search_candidate(client, [item("准备装修求推荐靠谱公司", "a"), item("新房装修实景案例", "b"), item(None, "c"), item("装修求助工长", "d")]) result = client.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修", "region": "", "max_selected": 2}) assert result.status_code == 201, result.text run = result.json() assert run["status"] == "completed" and run["analysis_mode"] == "rules" and run["model_connected"] is False assert run["candidate_digest"] == candidate["digest"] and run["search_task_id"] == created["id"] and run["validation_id"] assert run["conditions"] == {"service": "装修服务", "target_customer": "有装修需求的业主", "region": ""} assert [row["note_title"] for row in run["selected"]] == ["准备装修求推荐靠谱公司", "装修求助工长"] assert next(row for row in run["results"] if row["title"] == "新房装修实景案例")["classification"] == "peer_case" assert run["evidence"] and all(row["uncertainties"] for row in run["results"]) assert client.get("/api/discovery/runs/" + run["id"]).json() == run assert client.get("/api/discovery/runs").json()["items"][0] == run frozen = client.get("/api/learning/candidates/" + candidate["id"]).json() assert frozen["digest"] == candidate["digest"] and frozen["manifest"] == candidate["manifest"] service = client.app_reference.state.discovery owner = service.store.one("SELECT owner_id FROM discovery_runs WHERE id=$1", (run["id"],))["owner_id"] with pytest.raises(HTTPException) as error: service.get(run["id"], owner + "_other") assert error.value.status_code == 404 and service.list(owner + "_other") == [] for query in ("UPDATE discovery_runs SET data_json='{}' WHERE id=$1", "DELETE FROM discovery_runs WHERE id=$1"): with pytest.raises((StoreConflict, sqlite3.IntegrityError)): service.store.execute(query, (run["id"],)) if service.store.backend == "postgresql": with pytest.raises(StoreConflict): # Include dependent reading facts so PostgreSQL reaches the # immutable trigger instead of rejecting the FK dependency first. service.store.execute("TRUNCATE discovery_runs CASCADE") else: assert Store(service.store.home).one("SELECT id FROM discovery_runs WHERE id=$1", (run["id"],))["id"] == run["id"] with TestClient(client.app_reference, base_url=BASE) as anonymous: assert anonymous.get("/api/discovery/runs").status_code == 401 assert anonymous.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修"}).status_code in {401, 403} def test_discovery_failure_has_safe_persistent_record_and_strict_nonempty(discovery_client): client = discovery_client initialize(client) sample_items = [item("准备装修求推荐", "a")] created, candidate = search_candidate(client, sample_items) prefix = "/api/learning/candidates/" + candidate["id"] valid = client.post(prefix + "/verify", json={"expected_outputs": {"search_results": {"state": "nonempty"}}}) assert valid.status_code == 200 and valid.json()["status"] == "passed" for invalid in ({"state": "nonempty", "extra": True}, {"state": "other"}): assert client.post(prefix + "/verify", json={"expected_outputs": {"search_results": invalid}}).status_code == 422 sample_items.clear() empty = client.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修"}).json() assert empty["status"] == "search_failed" and empty["validation_id"] and not empty["selected"] assert client.get("/api/discovery/runs/" + empty["id"]).json() == empty sample_items.append(item("准备装修求推荐", "a")) previous = client.worker_reference.call async def foreign_source(method, params): value = await previous(method, params) if method == "verify": value["outputs"]["search_results"]["source_url"] = "https://example.test/search?token=private-fixture" return value client.worker_reference.call = foreign_source foreign = client.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修"}).json() assert foreign["status"] == "search_failed" and foreign["error_codes"] == ["DISCOVERY_SOURCE_MISMATCH"] assert "private-fixture" not in str(foreign) client.worker_reference.call = previous before = len([call for call in client.worker_reference.calls if call[0] == "verify"]) client.post("/api/learning/tasks/" + created["id"] + "/browser/close") failed = client.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修"}).json() assert failed["status"] == "failed" and failed["validation_id"] is None and failed["results"] == [] assert failed["error_codes"] == ["browser_session_required"] and failed["errors"] assert len([call for call in client.worker_reference.calls if call[0] == "verify"]) == before assert client.get("/api/discovery/runs/" + failed["id"]).json() == failed # Candidate shape/source mismatches never cause browser execution. service = client.app_reference.state.discovery for platform, entry in (("other", BASE), ("douyin", BASE), ("demo", "https://example.test"), ("xiaohongshu", BASE)): altered = copy.deepcopy(candidate) altered["manifest"].update(platform=platform, entry_url=entry) assert not service.supported_candidate(altered) calls = len([call for call in client.worker_reference.calls if call[0] == "verify"]) run_count = client.get("/api/discovery/runs").json()["total"] for invalid in ({"keyword": " "}, {"max_selected": True}, {"max_selected": 4}, {"region": None}, {"service": "二手车"}, {"target_customer": "二手车买家"}, {"service": "二手车", "target_customer": "二手车买家"}): assert client.post("/api/discovery/runs", json={"search_candidate_id": candidate["id"], "keyword": "装修", **invalid}).status_code == 422 assert len([call for call in client.worker_reference.calls if call[0] == "verify"]) == calls assert client.get("/api/discovery/runs").json()["total"] == run_count def test_discovery_nonempty_never_means_semantic_or_empty_success(): contract = {**EXTRACT, "limit": 20} def checks(value, current=contract): return output_assertion_checks({"search_results": value}, {"search_results": {"state": "nonempty"}}, {"search_results": current}) assert all(row["passed"] for row in checks(output([item(None, author=None)]))) for changed in (output([]), {**output([item("装修")]), "count": 2}, {**output([item("装修")]), "items": [{"title": "装修", "author": None, "url": None}]}, {**output([item("装修")]), "unexpected": True}): assert not all(row["passed"] for row in checks(changed)) assert not all(row["passed"] for row in checks([{"text": "装修"}], {"kind": "read"})) def test_discovery_rules_are_source_bound_deduplicated_and_do_not_infer_owners(): rows = [item("准备装修求推荐装修公司", "a"), item("装修完工求推荐沙发", "b"), item("装修真好看", "c"), item("好看", "d"), item("相同装修标题", "e"), item("相同装修标题", "f"), item("不应保存无效来源", "g", url="https://evil.test/explore/" + "g" * 24), item("新房装修", "a", url=BASE + "/search_result/" + "a" * 24 + "?token=private-fixture")] results, selected = select_results(rows, BASE + "/search", 3) assert len([row for row in results if row["note_id"] == "a" * 24]) == 1 assert next(row for row in results if row["note_id"] == "a" * 24)["classification"] == "ambiguous" assert all(row["classification"] == "ambiguous" and not row["selected"] for row in results if row["title"] == "相同装修标题") assert next(row for row in results if row["title"] == "装修完工求推荐沙发")["classification"] == "peer_case" assert next(row for row in results if row["title"] == "装修真好看")["classification"] == "related_content" assert next(row for row in results if row["title"] == "好看")["classification"] == "unrelated" assert next(row for row in results if row["title"] == "不应保存无效来源")["url"] is None assert all("?" not in row["note_url"] for row in selected) assert all(row["uncertainties"] and "真实装修需求尚未核实" in row["uncertainties"][0] for row in results) masked, chosen = select_results([item("装修求助[手机号已隐藏]", "a")], BASE, 1) assert not chosen and masked[0]["classification"] == "ambiguous" and masked[0]["url"] missing, chosen = select_results([item(None, "a"), item(" ", "b")], BASE, 2) assert not chosen and all(row["classification"] == "ambiguous" and row["url"] for row in missing) capped, chosen = select_results([item("准备装修求推荐", "a"), item("新房装修实景案例", "b"), item("装修经验分享材料流程", "c")], BASE, 1) assert [row["note_title"] for row in chosen] == ["准备装修求推荐"] assert all(not row["selected"] and any("达到本次保留数量上限" in reason for reason in row["reasons"]) for row in capped[1:]) assert not any("数量上限" in reason for reason in capped[0]["reasons"]) for word in ("完工", "竣工", "交付", "完装", "装完", "入住"): for negative in ("没", "未"): denied, chosen = select_results([item(f"装修还{negative}{word}，求推荐工长", "a")], BASE, 1) assert denied[0]["classification"] == "consumer_help" and chosen assert not any("已完工或入住" in reason for reason in denied[0]["reasons"])