test: cover v1 mock and integration boundaries

tests/fixtures/real_case_source/source_context.json

@@ -0,0 +1,60 @@
+{
+  "trace_id": "19b5b21d-a110-4dcd-9478-906a9752faa0",
+  "demand_content_id": "1",
+  "merge_leve2": "贪污腐败",
+  "name": "警花凌娅贪腐案",
+  "display_name": "基层公职人员贪腐警示案例",
+  "suggestion": null,
+  "score": null,
+  "dt": "2026-06-04",
+  "ext_data": {
+    "evidence_pack": {
+      "source_kind": "pattern_itemset",
+      "pattern_source_system": "mysql_topic_pattern",
+      "case_id_type": "post_id",
+      "source_post_id": "69509310",
+      "pattern_execution_id": "2038",
+      "mining_config_id": "2029",
+      "trace_id": "19b5b21d-a110-4dcd-9478-906a9752faa0",
+      "itemset_ids": ["391369", "391371"],
+      "itemset_items": [
+        {
+          "itemset_id": "391369",
+          "items": ["叙事结构", "综合性腐败", "表达手法"],
+          "support": 0.1,
+          "absolute_support": 5
+        },
+        {
+          "itemset_id": "391371",
+          "items": ["叙事结构", "基层公职人员", "表达手法"],
+          "support": 0.1,
+          "absolute_support": 5
+        }
+      ],
+      "category_bindings": [
+        {
+          "category_id": "corruption_warning",
+          "category_path": "社会议题 / 贪污腐败 / 警示案例"
+        }
+      ],
+      "element_bindings": [
+        {
+          "element_id": "grassroots_public_official",
+          "name": "基层公职人员"
+        }
+      ],
+      "matched_post_ids": [
+        "69509310",
+        "69509322",
+        "69509597",
+        "69510856",
+        "69514781",
+        "69535519",
+        "69509602"
+      ],
+      "seed_terms": ["叙事结构", "综合性腐败", "表达手法", "基层公职人员", "警花凌娅", "贪腐"],
+      "source_certainty": "db_validated",
+      "validation_status": "passed"
+    }
+  }
+}

tests/test_api.py

@@ -0,0 +1,60 @@
+from fastapi.testclient import TestClient
+
+from content_agent import api
+from content_agent.integrations.mock_platform import MockPlatformClient
+from content_agent.run_service import RunService
+
+
+def test_api_runs_and_queries_mock_chain(tmp_path, monkeypatch):
+    monkeypatch.setattr(api, "service", RunService(runtime_root=tmp_path / "runtime" / "v1"))
+    client = TestClient(api.app)
+
+    response = client.post("/runs", json={"trace_id": "test_trace_api", "platform": "douyin"})
+    assert response.status_code == 200
+    assert response.json()["platform_mode"] == "mock"
+
+    for path in [
+        "/runs/test_trace_api",
+        "/runs/test_trace_api/candidates",
+        "/runs/test_trace_api/rule-decisions",
+        "/runs/test_trace_api/source-edges",
+        "/runs/test_trace_api/final-output",
+        "/runs/test_trace_api/strategy-review",
+        "/runs/test_trace_api/validation",
+    ]:
+        get_response = client.get(path)
+        assert get_response.status_code == 200, path
+
+    review = client.get("/runs/test_trace_api/strategy-review").json()["data"]
+    assert review["summary"]["pool_count"] == 1
+    assert review["suggestions"]
+
+    validation = client.get("/runs/test_trace_api/validation").json()
+    assert validation["status"] == "pass"
+
+    summary = client.get("/runs/test_trace_api").json()
+    assert summary["validation_status"] == "pass"
+
+
+def test_api_accepts_real_platform_mode_without_changing_default(tmp_path, monkeypatch):
+    selected_modes = []
+
+    def fake_platform_client(self, platform_mode):
+        selected_modes.append(platform_mode)
+        return MockPlatformClient()
+
+    monkeypatch.setattr(RunService, "_platform_client", fake_platform_client)
+    monkeypatch.setattr(api, "service", RunService(runtime_root=tmp_path / "runtime" / "v1"))
+    client = TestClient(api.app)
+
+    default_response = client.post("/runs", json={"trace_id": "test_trace_default_mode"})
+    real_response = client.post(
+        "/runs",
+        json={"trace_id": "test_trace_real_mode", "platform_mode": "real"},
+    )
+
+    assert default_response.status_code == 200
+    assert default_response.json()["platform_mode"] == "mock"
+    assert real_response.status_code == 200
+    assert real_response.json()["platform_mode"] == "real"
+    assert selected_modes == ["mock", "real"]

tests/test_douyin_client.py

@@ -0,0 +1,233 @@
+import httpx
+import pytest
+
+from content_agent.integrations.douyin import CrawapiDouyinClient
+
+
+class FakeHttpClient:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.requests = []
+
+    def post(self, url, json, headers, timeout):
+        self.requests.append({"url": url, "json": json, "headers": headers, "timeout": timeout})
+        response = self.responses.pop(0)
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+
+def _response(status_code, data):
+    return httpx.Response(
+        status_code,
+        json=data,
+        request=httpx.Request("POST", "http://crawapi.test/endpoint"),
+    )
+
+
+def _client(responses):
+    return CrawapiDouyinClient(
+        base_url="http://crawapi.test",
+        keyword_path="/crawler/dou_yin/keyword",
+        content_portrait_path="/crawler/dou_yin/re_dian_bao/video_like_portrait",
+        default_account_id="771431222",
+        http_client=FakeHttpClient(responses),
+    )
+
+
+def test_douyin_keyword_search_maps_video_and_portrait_fields():
+    client = _client(
+        [
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": [
+                            {
+                                "aweme_id": "7615247738577423622",
+                                "desc": "早睡早起早上好",
+                                "author": {
+                                    "nickname": "让你心情变浅的兔子",
+                                    "sec_uid": "MS4wLjABAAAAdYc",
+                                },
+                                "statistics": {
+                                    "digg_count": 1931,
+                                    "comment_count": 45,
+                                    "share_count": 1968,
+                                    "collect_count": 462,
+                                },
+                                "text_extra": [{"hashtag_name": "早上好"}],
+                            }
+                        ],
+                        "has_more": True,
+                        "next_cursor": "10",
+                    }
+                },
+            ),
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": {
+                            "年龄": {
+                                "50+": {"percentage": "18.00%", "preference": "135.0"},
+                                "31-40": {"percentage": "20.00%", "preference": "80.0"},
+                            }
+                        }
+                    }
+                },
+            ),
+        ]
+    )
+
+    results = client.search(
+        {
+            "query_id": "q_001",
+            "query": "早上好祝福视频",
+            "origin_source": "pattern_itemset",
+        }
+    )
+
+    assert len(results) == 1
+    result = results[0]
+    assert result["candidate_id"] == "q_001_c_001"
+    assert result["aweme_id"] == "7615247738577423622"
+    assert result["author_sec_uid"] == "MS4wLjABAAAAdYc"
+    assert result["cha_list"] == ["#早上好"]
+    assert result["portrait_available"] is True
+    assert result["age_50_plus_level"] == "strong"
+    assert result["pattern_recall"] == "candidate_related"
+    assert result["candidate_relation"] == "derived_from_pattern_demand"
+    assert result["platform_auth_mode"] == "no_bearer"
+    assert client.http_client.requests[0]["json"]["account_id"] == "771431222"
+
+
+def test_douyin_keyword_search_returns_empty_list():
+    client = _client([_response(200, {"data": {"data": [], "has_more": False}})])
+
+    results = client.search(
+        {"query_id": "q_001", "query": "无结果", "origin_source": "pattern_itemset"}
+    )
+
+    assert results == []
+
+
+def test_douyin_keyword_search_http_error_is_sanitized():
+    client = _client([_response(500, {"error": "server failed"})])
+
+    with pytest.raises(RuntimeError, match="keyword_search failed: HTTP 500"):
+        client.search(
+            {"query_id": "q_001", "query": "接口失败", "origin_source": "pattern_itemset"}
+        )
+
+
+def test_douyin_portrait_http_error_marks_candidate_missing_portrait():
+    client = _client(
+        [
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": [
+                            {
+                                "aweme_id": "7615247738577423622",
+                                "desc": "早上好",
+                                "author": {"nickname": "作者", "sec_uid": "MS4wLjABAAAA001"},
+                                "statistics": {"digg_count": 1},
+                            }
+                        ]
+                    }
+                },
+            ),
+            _response(500, {"error": "portrait failed"}),
+        ]
+    )
+
+    result = client.search(
+        {
+            "query_id": "q_001",
+            "query": "早上好",
+            "origin_source": "pattern_itemset",
+        }
+    )[0]
+
+    assert result["portrait_available"] is False
+    assert result["age_50_plus_level"] == "missing"
+
+
+def test_douyin_keyword_search_can_limit_results_per_query():
+    client = CrawapiDouyinClient(
+        base_url="http://crawapi.test",
+        keyword_path="/crawler/dou_yin/keyword",
+        content_portrait_path="/crawler/dou_yin/re_dian_bao/video_like_portrait",
+        max_results_per_query=1,
+        http_client=FakeHttpClient(
+            [
+                _response(
+                    200,
+                    {
+                        "data": {
+                            "data": [
+                                {"aweme_id": "1", "author": {}, "statistics": {}},
+                                {"aweme_id": "2", "author": {}, "statistics": {}},
+                            ]
+                        }
+                    },
+                ),
+                _response(200, {"data": {"data": {"年龄": {}}}}),
+            ]
+        ),
+    )
+
+    results = client.search(
+        {"query_id": "q_001", "query": "限量", "origin_source": "pattern_itemset"}
+    )
+
+    assert [result["aweme_id"] for result in results] == ["1"]
+
+
+def test_douyin_portrait_supports_dimensions_shape_and_excludes_41_to_50():
+    client = _client(
+        [
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": [
+                            {
+                                "aweme_id": "7635992906608060495",
+                                "desc": "高考加油",
+                                "author": {"nickname": "一个富贵", "sec_uid": "MS4wLjABAAAA001"},
+                                "statistics": {"digg_count": 100},
+                            }
+                        ]
+                    }
+                },
+            ),
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": {
+                            "dimensions": {
+                                "年龄": [
+                                    {"name": "41-50", "percentage": "30.00%", "preference": "150.0"},
+                                    {"name": "50-", "percentage": "4.83%", "preference": "13.80"},
+                                ]
+                            }
+                        }
+                    }
+                },
+            ),
+        ]
+    )
+
+    result = client.search(
+        {"query_id": "q_001", "query": "高考加油", "origin_source": "pattern_itemset"}
+    )[0]
+
+    assert result["portrait_available"] is True
+    assert result["age_50_plus_ratio"] == 0.0483
+    assert result["age_50_plus_level"] == "weak"
+    assert result["age_distribution"][0]["is_50_plus"] is False
+    assert result["age_distribution"][1]["is_50_plus"] is True

tests/test_platform_access.py

@@ -0,0 +1,34 @@
+from content_agent.business_modules import platform_access
+
+
+class DuplicateVideoClient:
+    def search(self, query):
+        return [
+            {
+                "candidate_id": f"{query['query_id']}_c_001",
+                "query_id": query["query_id"],
+                "aweme_id": "7601814454925298994",
+                "desc": "重复视频",
+            }
+        ]
+
+
+def test_platform_access_deduplicates_same_aweme_across_queries():
+    queries = [
+        {
+            "query_id": "q_001",
+            "query": "基层公职人员贪腐案例",
+            "generation_type": "item_combo",
+        },
+        {
+            "query_id": "q_002",
+            "query": "综合性腐败 警示案例",
+            "generation_type": "light_extension",
+        },
+    ]
+
+    results = platform_access.run(queries, DuplicateVideoClient())
+
+    assert len(results) == 1
+    assert results[0]["query_id"] == "q_001"
+    assert results[0]["query"] == "基层公职人员贪腐案例"

tests/test_rule_pack_reading.py

@@ -0,0 +1,46 @@
+from copy import deepcopy
+
+from content_agent.business_modules.rule_judgment.evaluator import decide
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+def test_rule_pack_thresholds_drive_decision(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    state = service.start_run(RunStartRequest(trace_id="test_trace_rule_threshold"))
+
+    policy_bundle = deepcopy(state["policy_bundle"])
+    thresholds = policy_bundle["rule_pack"]["thresholds"]
+    thresholds[0]["min_score"] = 80
+    thresholds[1]["min_score"] = 70
+    thresholds[1]["max_score"] = 79
+
+    decision = decide("test_trace_rule_threshold", 1, state["evidence_bundles"][0], policy_bundle)
+    assert decision["final_action"] == "CANDIDATE"
+    assert decision["reason_code"] == "video_score_candidate"
+    assert decision["replay_fields"]["matched_threshold"] == "70<=score<=79"
+
+
+def test_rule_pack_hard_gate_reason_code_drives_decision(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    state = service.start_run(RunStartRequest(trace_id="test_trace_rule_gate"))
+
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["source_evidence"] = {}
+
+    decision = decide("test_trace_rule_gate", 1, bundle, state["policy_bundle"])
+    assert decision["final_action"] == "REJECT"
+    assert decision["reason_code"] == "missing_source_evidence"
+    assert decision["matched_hard_gates"] == ["missing_source_evidence"]
+
+
+def test_missing_score_fallback_uses_rule_pack_threshold(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    state = service.start_run(RunStartRequest(trace_id="test_trace_score_fallback"))
+
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["relevance_signal"]["score"] = None
+
+    decision = decide("test_trace_score_fallback", 1, bundle, state["policy_bundle"])
+    assert decision["final_action"] == "REJECT"
+    assert decision["reason_code"] == "video_score_reject"

tests/test_runtime_files.py

@@ -0,0 +1,154 @@
+import json
+from pathlib import Path
+
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+def test_runtime_files_are_parseable_and_consistent(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_files"))
+    run_dir = service.runtime.run_dir("test_trace_files")
+
+    for json_file in ["source_context.json", "pattern_seed_pack.json", "final_output.json"]:
+        json.loads((run_dir / json_file).read_text(encoding="utf-8"))
+
+    for jsonl_file in [
+        "queries.jsonl",
+        "candidate_pool.jsonl",
+        "media_assets.jsonl",
+        "rule_decisions.jsonl",
+        "trace_events.jsonl",
+        "source_edges.jsonl",
+        "search_clues.jsonl",
+    ]:
+        for line in (run_dir / jsonl_file).read_text(encoding="utf-8").splitlines():
+            json.loads(line)
+
+    candidates = service.read_jsonl("test_trace_files", "candidate_pool.jsonl")
+    decisions = service.read_jsonl("test_trace_files", "rule_decisions.jsonl")
+    edges = service.read_jsonl("test_trace_files", "source_edges.jsonl")
+    final_output = service.read_json("test_trace_files", "final_output.json")
+    source_context = service.read_json("test_trace_files", "source_context.json")
+
+    candidate_aweme_ids = {candidate["aweme_id"] for candidate in candidates}
+    decision_entity_ids = {decision["entity_id"] for decision in decisions}
+    assert candidate_aweme_ids == decision_entity_ids
+
+    assert {decision["final_action"] for decision in decisions} <= {
+        "POOL",
+        "CANDIDATE",
+        "PENDING",
+        "REJECT",
+    }
+
+    edge_ids = {edge["edge_id"] for edge in edges}
+    for asset in final_output["content_assets"]:
+        assert set(asset["source_edge_ids"]) <= edge_ids
+
+    assert source_context["ext_data"]["evidence_pack"]["pattern_execution_id"] == "2038"
+
+    validation = service.validate_run("test_trace_files")
+    assert validation["status"] == "pass"
+
+
+def test_runtime_validation_catches_summary_drift(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_drift"))
+
+    final_output_path = service.runtime.run_dir("test_trace_drift") / "final_output.json"
+    final_output = json.loads(final_output_path.read_text(encoding="utf-8"))
+    final_output["summary"]["pool_count"] = 99
+    final_output_path.write_text(
+        json.dumps(final_output, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+    validation = service.validate_run("test_trace_drift")
+    assert validation["status"] == "fail"
+    assert any(finding["check_id"] == "summary_mismatch" for finding in validation["findings"])
+
+
+def test_runtime_validation_catches_missing_final_decision_record(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_missing_decision_record"))
+
+    final_output_path = service.runtime.run_dir("test_trace_missing_decision_record") / "final_output.json"
+    final_output = json.loads(final_output_path.read_text(encoding="utf-8"))
+    final_output["decision_records"] = final_output["decision_records"][:-1]
+    final_output_path.write_text(
+        json.dumps(final_output, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+    validation = service.validate_run("test_trace_missing_decision_record")
+    assert validation["status"] == "fail"
+    assert any(finding["check_id"] == "final_decision_missing" for finding in validation["findings"])
+
+
+def test_runtime_validation_catches_aweme_id_source_pollution(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_aweme_pollution"))
+
+    final_output_path = service.runtime.run_dir("test_trace_aweme_pollution") / "final_output.json"
+    final_output = json.loads(final_output_path.read_text(encoding="utf-8"))
+    source_evidence = final_output["decision_records"][0]["source_evidence"]
+    source_evidence["source_post_id"] = source_evidence["candidate_aweme_id"]
+    final_output_path.write_text(
+        json.dumps(final_output, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+    validation = service.validate_run("test_trace_aweme_pollution")
+    assert validation["status"] == "fail"
+    assert any(
+        finding["check_id"] == "source_evidence_aweme_pollution"
+        for finding in validation["findings"]
+    )
+
+
+def test_runtime_validation_catches_reject_source_path_break(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_reject_path_break"))
+
+    edges_path = service.runtime.run_dir("test_trace_reject_path_break") / "source_edges.jsonl"
+    edges = [
+        json.loads(line)
+        for line in edges_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    edges = [
+        edge
+        for edge in edges
+        if not (
+            edge.get("edge_type") == "query_to_video"
+            and edge.get("to_node_id") == "7390000000000000099"
+        )
+    ]
+    edges_path.write_text(
+        "".join(json.dumps(edge, ensure_ascii=False, separators=(",", ":")) + "\n" for edge in edges),
+        encoding="utf-8",
+    )
+
+    validation = service.validate_run("test_trace_reject_path_break")
+    assert validation["status"] == "fail"
+    assert any(finding["check_id"] == "source_path_broken" for finding in validation["findings"])
+
+
+def test_real_source_fixture_keeps_upstream_evidence_pack(tmp_path):
+    source_path = Path("tests/fixtures/real_case_source/source_context.json")
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_real_source", source=str(source_path)))
+
+    source_context = service.read_json("test_trace_real_source", "source_context.json")
+    evidence_pack = source_context["ext_data"]["evidence_pack"]
+    decisions = service.read_jsonl("test_trace_real_source", "rule_decisions.jsonl")
+
+    assert evidence_pack["pattern_source_system"] == "mysql_topic_pattern"
+    assert evidence_pack["source_certainty"] == "db_validated"
+    assert evidence_pack["validation_status"] == "passed"
+    assert evidence_pack["source_post_id"] == "69509310"
+    assert evidence_pack["upstream_trace_id"] == "19b5b21d-a110-4dcd-9478-906a9752faa0"
+    assert decisions[0]["source_evidence"]["source_certainty"] == "db_validated"
+    assert decisions[0]["source_evidence"]["candidate_aweme_id"] not in evidence_pack["matched_post_ids"]
+    assert service.validate_run("test_trace_real_source")["status"] == "pass"

tests/test_source_evidence.py

@@ -0,0 +1,40 @@
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+def test_source_evidence_inherits_evidence_pack_without_rewriting_origin(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    service.start_run(RunStartRequest(trace_id="test_trace_source_evidence"))
+
+    source_context = service.read_json("test_trace_source_evidence", "source_context.json")
+    evidence_pack = source_context["ext_data"]["evidence_pack"]
+    decisions = service.read_jsonl("test_trace_source_evidence", "rule_decisions.jsonl")
+    final_output = service.read_json("test_trace_source_evidence", "final_output.json")
+
+    source_evidence = decisions[0]["source_evidence"]
+    for field in [
+        "source_kind",
+        "pattern_source_system",
+        "case_id_type",
+        "source_post_id",
+        "pattern_execution_id",
+        "mining_config_id",
+        "itemset_ids",
+        "itemset_items",
+        "category_bindings",
+        "element_bindings",
+        "matched_post_ids",
+        "seed_terms",
+        "trace_id",
+        "source_certainty",
+        "validation_status",
+    ]:
+        assert source_evidence[field] == evidence_pack[field]
+
+    assert source_evidence["candidate_aweme_id"] != source_evidence["source_post_id"]
+    assert source_evidence["candidate_aweme_id"] not in source_evidence["matched_post_ids"]
+    assert source_evidence["candidate_relation"] == "mock_pattern_matched"
+    assert final_output["content_assets"][0]["source_evidence"]["source_edge_ids"]
+    assert {
+        record["decision_id"] for record in final_output["decision_records"]
+    } == {"d_001", "d_002", "d_003"}

tests/test_v1_graph.py

@@ -0,0 +1,18 @@
+from content_agent.integrations.runtime_files import RUNTIME_FILENAMES
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+def test_v1_graph_generates_all_runtime_files(tmp_path):
+    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
+    state = service.start_run(RunStartRequest(trace_id="test_trace_graph"))
+
+    assert state["status"] == "success"
+    status = service.runtime.file_status("test_trace_graph")
+    assert all(status[name] for name in RUNTIME_FILENAMES)
+
+    final_output = service.read_json("test_trace_graph", "final_output.json")
+    assert final_output["summary"]["pool_count"] == 1
+    assert final_output["summary"]["pending_count"] == 1
+    assert final_output["summary"]["reject_count"] == 1
+