| 1234567891011121314151617181920212223242526272829 |
- """失败的游走 query 不消耗预算(标签/作者跳失败都不扣边预算 + 不累计总闸)。"""
- from content_agent.business_modules.walk_engine import run_bounded_walk
- from content_agent.integrations.walk_graph_json import WalkGraphStore
- from tests.p6_walk_helpers import FakeWalkPlatformClient, build_initial_walk_context, set_v4_allow_walk
- def test_failed_tag_walk_does_not_consume_budget(tmp_path):
- context = build_initial_walk_context(tmp_path)
- set_v4_allow_walk(context["rule_decisions"][0], True)
- # fail_tag:所有标签搜索都失败;作者游走仍成功(其 raw_payload 带 budget 快照)。
- result = run_bounded_walk(platform_client=FakeWalkPlatformClient(fail_tag=True), **context)
- tag_failed = [
- row for row in result["walk_actions"]
- if row["edge_id"] == "hashtag_to_query" and row["walk_status"] == "failed"
- ]
- assert tag_failed, "应有失败的标签游走"
- author_success = [
- row for row in result["walk_actions"]
- if row["edge_id"] == "author_to_works" and row["walk_status"] == "success"
- ]
- assert author_success, "作者游走应成功并带 budget 快照"
- tag_cap = WalkGraphStore().load_policy()["edge_budgets_by_id"]["hashtag_to_query"]["max_total_actions"]
- snap = author_success[0]["raw_payload"]["remaining_budget_snapshot"]
- # 标签全失败 → 一次都不扣 → 标签预算仍满额(failed 不消耗)。
- assert snap["hashtag_to_query"] == tag_cap
|
