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

"""V3-M4A: WalkGraphStore 读三个游走配置 JSON + 校验 + 拍板值解包。"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from content_agent.integrations.walk_graph_json import (
    EDGE_CATALOG_PATH,
    PROFILE_EDGE_STATUSES,
    WalkGraphStore,
    _validate_graph,
    _validate_profile,
    edge_budgets_for_platform,
    edge_permission,
    edge_supported,
    low_budget,
)

PROFILE_DIR = Path("tech_documents/数据接口与来源/platform_profiles")


def test_graph_has_8_nodes_and_9_edges():
    # M8:删 query_next_page 边后 10→9;search_page 节点保留(孤儿不报错)。
    graph = WalkGraphStore().load_graph()
    assert len(graph["nodes"]) == 8
    assert len(graph["edges"]) == 9
    assert "query_next_page" not in {edge["edge_id"] for edge in graph["edges"]}
    assert {edge["gate"] for edge in graph["edges"]} == {"none", "decision_gated", "keep_only"}


def test_graph_edges_match_platform_neutral_catalog():
    graph = WalkGraphStore().load_graph()
    catalog = json.loads(EDGE_CATALOG_PATH.read_text(encoding="utf-8"))

    assert {edge["edge_id"] for edge in graph["edges"]} == {
        row["edge_id"] for row in catalog["walk_edge_catalog"]
    }


def test_policy_unwraps_pinned_values():
    policy = WalkGraphStore().load_policy()
    # 2026-06-24 拍板:总闸收敛到 10、删 max_reseed_rounds;deny / halve_min_1(max(1, budget//2))。
    assert policy["global"]["max_total_actions_per_run"] == 10
    assert policy["global"]["max_depth"] == 5
    assert "max_reseed_rounds" not in policy["global"]
    assert policy["edge_permissions"]["KEEP_CONTENT_FOR_REVIEW"]["video_to_hashtag"] == "deny"
    assert policy["budget_tiers"]["low_budget"] == "halve_min_1"
    assert low_budget(3) == 1
    assert low_budget(10) == 5


def test_policy_pins_gemini_workers_without_call_cap():
    # Gemini 数量上限已取消;仅保留判定并发度。M8 删 max_reseed_rounds。
    policy_global = WalkGraphStore().load_policy()["global"]
    assert set(policy_global) == {
        "max_total_actions_per_run",
        "max_depth",
        "gemini_max_workers",
    }
    assert policy_global["gemini_max_workers"] == 5


def test_policy_allows_smoke_env_overrides(monkeypatch):
    monkeypatch.setenv("CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN", "4")
    monkeypatch.setenv("CONTENT_AGENT_WALK_MAX_DEPTH", "2")
    monkeypatch.setenv("CONTENT_AGENT_GEMINI_MAX_WORKERS", "3")

    policy_global = WalkGraphStore().load_policy()["global"]

    assert policy_global["max_total_actions_per_run"] == 4
    assert policy_global["max_depth"] == 2
    assert policy_global["gemini_max_workers"] == 3
    assert set(policy_global) == {
        "max_total_actions_per_run",
        "max_depth",
        "gemini_max_workers",
    }


def test_policy_edge_budgets_match_decided_values():
    # 2026-06-24 拍板:标签 5、作者 5、每作者作品 5;query_next_page 已删。
    budgets = WalkGraphStore().load_policy()["edge_budgets_by_id"]
    assert "query_next_page" not in budgets
    assert budgets["author_to_works"]["max_total_actions"] == 5
    assert budgets["author_to_works"]["max_works_per_author"] == 5
    assert budgets["hashtag_to_query"]["max_total_actions"] == 5


def test_platform_edge_budgets_migrate_default_and_kuaishou_author_budget():
    policy = WalkGraphStore().load_policy()
    rows = policy["platform_edge_budgets_by_platform_edge"]

    assert rows[("default", "__global__")]["max_total_actions"] == 10
    assert rows[("default", "__global__")]["max_depth"] == 5
    assert rows[("default", "author_to_works")]["max_total_actions"] == 5
    assert rows[("kuaishou", "author_to_works")]["max_works_per_author"] == 5

    kuaishou = edge_budgets_for_platform(policy, "kuaishou")
    assert kuaishou["author_to_works"]["max_total_actions"] == 5
    assert kuaishou["author_to_works"]["operator_display_status"] == "budget_blocked"


def test_three_short_video_author_edges_match_m3_contract():
    store = WalkGraphStore()
    assert edge_supported(store.load_profile("douyin"), "author_to_works")
    assert edge_supported(store.load_profile("douyin"), "author_work_to_content")
    assert edge_supported(store.load_profile("kuaishou"), "author_to_works")
    assert edge_supported(store.load_profile("kuaishou"), "author_work_to_content")
    profile = store.load_profile("shipinhao")
    assert not edge_supported(profile, "author_to_works")
    assert not edge_supported(profile, "author_work_to_content")
    assert edge_supported(profile, "hashtag_to_query")
    # 终端边不在 profile(平台无关控制边)→ 恒 supported。
    assert edge_supported(profile, "decision_to_asset")


def test_profile_edge_status_contract_accepts_four_states():
    assert PROFILE_EDGE_STATUSES == {"supported", "blocked", "no_interface", "planned"}
    edge_ids = {"author_to_works", "video_to_hashtag"}
    findings = _validate_profile(
        {
            "edges": {
                "author_to_works": {"status": "supported"},
                "video_to_hashtag": {"status": "no_interface"},
            }
        },
        edge_ids,
    )
    assert findings == []

    bad_status = _validate_profile({"edges": {"author_to_works": {"status": "weak"}}}, edge_ids)
    assert any(finding["check_id"] == "profile_edge_status" for finding in bad_status)

    bad_shape = _validate_profile({"edges": {"_all": "blocked"}}, edge_ids)
    assert any(finding["check_id"] == "profile_edge_ref" for finding in bad_shape)
    assert any(finding["check_id"] == "profile_edge_invalid" for finding in bad_shape)


def test_edge_supported_only_allows_supported_status():
    assert edge_supported({"edges": {"edge": {"status": "supported"}}}, "edge")
    for status in ["blocked", "no_interface", "planned"]:
        assert not edge_supported({"edges": {"edge": {"status": status}}}, "edge")
    assert edge_supported({"edges": {}}, "decision_to_asset")


def test_all_platform_profiles_load_and_blocked_profiles_are_explicit():
    store = WalkGraphStore()
    graph_edge_ids = {edge["edge_id"] for edge in store.load_graph()["edges"]}

    for path in sorted(PROFILE_DIR.glob("*.json")):
        profile = store.load_profile(path.stem)
        assert "_all" not in profile["edges"]
        if path.stem in {"toutiao", "zhihu"}:
            assert set(profile["edges"]) == graph_edge_ids
            assert {row["status"] for row in profile["edges"].values()} == {"blocked"}


def test_edge_permission_defaults_to_deny():
    policy = WalkGraphStore().load_policy()
    assert edge_permission(policy, "ADD_TO_CONTENT_POOL", "author_to_works") == "allow"
    assert edge_permission(policy, "KEEP_CONTENT_FOR_REVIEW", "author_to_works") == "allow_low_budget"
    assert edge_permission(policy, "REJECT_CONTENT", "author_to_works") == "deny"
    assert edge_permission(policy, None, "author_to_works") == "deny"
    assert edge_permission(policy, "ADD_TO_CONTENT_POOL", "unknown_edge") == "deny"


def test_unknown_graph_edge_fails_validation():
    graph = WalkGraphStore().load_graph()
    tampered = {**graph, "edges": [*graph["edges"], {**graph["edges"][0], "edge_id": "made_up_edge"}]}
    findings = _validate_graph(tampered, {edge["edge_id"] for edge in graph["edges"]})
    assert any(finding["check_id"] == "edge_not_in_catalog" for finding in findings)
    with pytest.raises(ValueError):
        store = WalkGraphStore()
        bad_policy = {**store.load_policy(), "edge_budgets": [{"edge_id": "made_up_edge"}]}
        from content_agent.integrations.walk_graph_json import _raise_on_fail, _validate_policy

        _raise_on_fail(_validate_policy(bad_policy, {e["edge_id"] for e in graph["edges"]}), "walk policy")