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

from content_agent.business_modules import search_intent, source_seed
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE, real_source_payload


def test_v4_m1_direct_query_sources_replay_from_seed_pack(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run("run_1")
    seed = source_seed.run("run_1", "policy_1", real_source_payload(), runtime)
    client = FakeQueryVariantClient({"爱国情感": "不应该调用"})

    queries = search_intent.run(
        "run_1",
        "policy_1",
        seed["pattern_seed_pack"],
        runtime,
        client,
        strategy_version="V4",
    )

    assert client.calls == []
    # 首轮搜索池三路直连:seed_terms + query_seed_points + 分类叶子元素,不调 LLM 变体。
    methods = {row["search_query_generation_method"] for row in queries}
    assert {"seed_term", "query_seed_point", "category_leaf_element"} <= methods
    assert "llm_variant" not in methods
    assert all("llm_variant_of" not in row for row in queries)
    assert all(row["pattern_seed_ref"]["query_source_text"] == row["search_query"] for row in queries)
    assert all(row["pattern_seed_ref"]["query_source_rank"] >= 1 for row in queries)
    assert all(row["raw_payload"]["query_source_refs"] for row in queries)
    source_types = {row["pattern_seed_ref"]["query_source_type"] for row in queries}
    assert {"seed_term", "query_seed_point", "category_terminal_element"} <= source_types


def test_v4_m1_run_service_does_not_call_query_variant_client(tmp_path):
    client = FakeQueryVariantClient({"爱国情感": "不应该调用"})
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=client,
    )

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

    queries = service.read_jsonl(state["run_id"], "search_queries.jsonl")
    assert state["status"] == "success"
    assert client.calls == []
    assert queries
    assert "llm_variant" not in {row["search_query_generation_method"] for row in queries}