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

import re
from pathlib import Path

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


CURRENT_CONTRACT_ROOTS = [
    Path("content_agent"),
    Path("tests"),
    Path("sql"),
    Path("product_documents"),
]


def test_current_contract_has_no_deprecated_content_action_or_old_status():
    text = _joined_text(CURRENT_CONTRACT_ROOTS)

    assert "HOLD_CONTENT_" + "PENDING" not in text
    assert "weak_" + "effective" not in text
    old_blocked_effect = re.compile(
        r"[\"']?search_query_effect_status[\"']?\s*[:=]\s*[\"']blocked[\"']"
    )
    assert old_blocked_effect.search(text) is None


def test_current_contract_has_no_old_walk_strategy_source_or_id():
    text = _joined_text(CURRENT_CONTRACT_ROOTS)

    assert "douyin_available_" + "walk_strategy_v1" not in text
    assert "douyin_available_" + "walk_strategy.v1.json" not in text
    old_path = Path("product_documents/抖音游走策略") / (
        "douyin_available_" + "walk_strategy.v1.json"
    )
    assert not old_path.exists()


def test_current_code_does_not_hardcode_completeness_true():
    text = _joined_text([Path("content_agent")])

    assert "run_path_complete" + ": True" not in text
    assert "run_path_complete" + " = True" not in text
    assert "trace_complete" + ": True" not in text
    assert "trace_complete" + " = True" not in text


def test_new_run_does_not_emit_old_strategy_name(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )

    final_output = service.read_json(state["run_id"], "final_output.json")
    text = str(final_output)
    assert "douyin_available_" + "walk_strategy_v1" not in text
    assert final_output["walk_strategy"]["walk_strategy_id"] == "douyin_walk_strategy_v1"


def _joined_text(roots: list[Path]) -> str:
    chunks = []
    for root in roots:
        for path in root.rglob("*"):
            if path.is_file() and path.suffix in {".py", ".sql", ".md", ".json"}:
                chunks.append(path.read_text(encoding="utf-8"))
    return "\n".join(chunks)