"""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")