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

import json
from copy import deepcopy
from pathlib import Path

from content_agent.integrations.walk_strategy_json import (
    EDGE_CATALOG_PATH,
    REQUIRED_SECTIONS,
    WalkStrategyStore,
    validate_walk_strategy_config,
)


CONFIG_PATH = Path("product_documents/抖音游走策略/douyin_walk_strategy.v1.json")


def test_walk_strategy_config_loads_only_p6_runtime_source():
    store = WalkStrategyStore(Path("."))

    strategy = store.load_walk_strategy()

    assert strategy["strategy_id"] == "douyin_walk_strategy_v1"
    assert strategy["walk_strategy_version"] == "V1.0"
    assert strategy["walk_strategy_source_ref"]["file"] == str(CONFIG_PATH)
    assert strategy["source_of_truth"] == str(CONFIG_PATH)


def test_walk_strategy_config_sections_and_references_are_closed():
    strategy = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))

    findings = validate_walk_strategy_config(strategy, root_dir=Path("."))

    assert findings == []
    assert [section for section in REQUIRED_SECTIONS if section not in strategy] == []


def test_walk_strategy_config_keeps_only_consumed_sections():
    # V5 M1: 边目录改由平台无关 catalog 校验;strategy 内旧段暂留为兼容副本。
    strategy = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))

    assert set(REQUIRED_SECTIONS) == {
        "walk_rule_pack_binding",
        "v4_walk_gate",
        "walk_fact_contract",
    }
    assert [section for section in REQUIRED_SECTIONS if section not in strategy] == []
    assert "walk_edge_catalog" not in REQUIRED_SECTIONS


def test_walk_strategy_validation_uses_platform_neutral_edge_catalog():
    strategy = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    catalog = json.loads(EDGE_CATALOG_PATH.read_text(encoding="utf-8"))

    assert "walk_edge_catalog" not in strategy
    assert validate_walk_strategy_config(strategy, root_dir=Path(".")) == []
    assert {row["edge_id"] for row in catalog["walk_edge_catalog"]} == {
        "search_page_to_content",
        "video_to_author",
        "author_to_works",
        "author_work_to_content",
        "video_to_hashtag",
        "hashtag_to_query",
        "decision_to_asset",
        "path_stop",
        "budget_downgrade",
    }


def test_walk_strategy_rejects_unknown_binding_and_gate_edges():
    strategy = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))

    bad_binding = deepcopy(strategy)
    bad_binding["walk_rule_pack_binding"][0]["edge_id"] = "made_up_edge"
    binding_findings = validate_walk_strategy_config(bad_binding, root_dir=Path("."))
    assert any(finding["check_id"] == "edge_ref" for finding in binding_findings)

    bad_gate = deepcopy(strategy)
    bad_gate["v4_walk_gate"][0]["applies_to_edges"] = ["author_to_works", "made_up_edge"]
    gate_findings = validate_walk_strategy_config(bad_gate, root_dir=Path("."))
    assert any(finding["check_id"] == "v4_walk_gate_edge_ref" for finding in gate_findings)


def test_walk_strategy_config_uses_clue_id_and_real_rule_packs():
    strategy = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    facts = {row["runtime_file"]: row for row in strategy["walk_fact_contract"]}

    assert facts["search_clues.jsonl"]["unique_key"] == [
        "run_id",
        "policy_run_id",
        "clue_id",
    ]
    # M4 砍包受控变化:4 future 包及其 7 条 binding 已物理删除,仅剩内容包归属。
    assert {
        (row["rule_pack_id"], row["rule_pack_version"])
        for row in strategy["walk_rule_pack_binding"]
    } == {
        ("douyin_content_discovery_rule_pack_v1", "4.0.0"),
    }


def test_walk_strategy_binding_is_loaded_from_walk_strategy():
    from content_agent.business_modules.walk_engine import _binding_by_edge_id
    from content_agent.integrations.walk_strategy_json import WalkStrategyStore

    walk_strategy = WalkStrategyStore().load_walk_strategy()
    binding_by_edge = _binding_by_edge_id(walk_strategy)

    # M4 砍包受控变化:仅 decision_to_asset 保留 content binding,其余边走 fallback 戳。
    assert set(binding_by_edge) == {"decision_to_asset"}
    assert binding_by_edge["decision_to_asset"]["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"