content-find-agent-new/tests/test_v4_m6_e2e_acceptance.py

from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.gemini_helpers import FakeGeminiVideoClient
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE


def test_v4_m6_controlled_e2e_builds_complete_runtime_without_external_clients(tmp_path):
    query_client = FakeQueryVariantClient({"爱国情感": "不应该调用"})
    gemini_client = FakeGeminiVideoClient()
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=query_client,
        gemini_video_client=gemini_client,
    )

    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V4")
    )
    run_id = state["run_id"]

    assert state["status"] == "success"
    assert query_client.calls == []
    assert gemini_client.calls

    search_queries = service.read_jsonl(run_id, "search_queries.jsonl")
    content_items = service.read_jsonl(run_id, "discovered_content_items.jsonl")
    media_records = service.read_jsonl(run_id, "content_media_records.jsonl")
    recall_evidence = service.read_jsonl(run_id, "pattern_recall_evidence.jsonl")
    rule_decisions = service.read_jsonl(run_id, "rule_decisions.jsonl")
    walk_actions = service.read_jsonl(run_id, "walk_actions.jsonl")
    source_paths = service.read_jsonl(run_id, "source_path_records.jsonl")
    search_clues = service.read_jsonl(run_id, "search_clues.jsonl")
    final_output = service.read_json(run_id, "final_output.json")
    strategy_review = service.read_json(run_id, "strategy_review.json")

    assert {"query_seed_point"} <= {
        row["search_query_generation_method"] for row in search_queries
    }
    assert "llm_variant" not in {row["search_query_generation_method"] for row in search_queries}
    assert content_items
    assert media_records
    assert recall_evidence
    assert rule_decisions
    assert walk_actions
    assert source_paths
    assert search_clues
    assert final_output["policy"]["strategy_version"] == "V4"
    assert strategy_review["schema_version"] == "runtime_record.v1"
    assert all(row["scorecard"]["schema_version"] == "v4_scorecard.v1" for row in rule_decisions)
    assert all("allow_walk" in row["decision_replay_data"] for row in rule_decisions)
    assert any(path["source_path_type"] == "pattern_to_search_query" for path in source_paths)

    validation = service.validate_run(run_id)
    assert validation["status"] == "pass", validation["findings"]