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

"""M9E:Gate 1 需求级终端元素准入(非抖音 only,只读 PG 判叶子,PG client 用 fake)。"""

from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest


class FakePatternPg:
    def __init__(self, has_terminal: bool) -> None:
        self.has_terminal = has_terminal
        self.calls: list[tuple] = []

    def has_terminal_element(self, execution_id, category_ids):
        self.calls.append((execution_id, list(category_ids)))
        return self.has_terminal


def _source(category_ids):
    return {
        "ext_data": {
            "evidence_pack": {
                "pattern_execution_id": 581,
                "itemset_items": [{"category_id": cid} for cid in category_ids],
            }
        }
    }


def _svc(has_terminal):
    return RunService(pattern_pg=FakePatternPg(has_terminal))


def _req(platform, mode="real"):
    return RunStartRequest(platform=platform, platform_mode=mode)


def test_gate1_blocks_non_douyin_without_terminal_element():
    svc = _svc(has_terminal=False)
    assert svc._gate1_block_reason(_req("kuaishou"), _source([1, 2])) == "gate1_no_terminal_element"


def test_gate1_allows_non_douyin_with_terminal_element():
    svc = _svc(has_terminal=True)
    assert svc._gate1_block_reason(_req("shipinhao"), _source([1, 2])) is None


def test_gate1_skips_douyin():
    pg = FakePatternPg(has_terminal=False)
    svc = RunService(pattern_pg=pg)
    assert svc._gate1_block_reason(_req("douyin"), _source([1, 2])) is None
    assert pg.calls == []  # 抖音不查 PG


def test_gate1_skips_mock_mode():
    pg = FakePatternPg(has_terminal=False)
    svc = RunService(pattern_pg=pg)
    assert svc._gate1_block_reason(_req("kuaishou", mode="mock"), _source([1, 2])) is None
    assert pg.calls == []


def test_gate1_skips_when_no_category_ids():
    pg = FakePatternPg(has_terminal=False)
    svc = RunService(pattern_pg=pg)
    assert svc._gate1_block_reason(_req("kuaishou"), _source([])) is None
    assert pg.calls == []


def test_gate1_blocked_state_shape(tmp_path):
    svc = RunService(runtime_root=tmp_path / "runtime", pattern_pg=FakePatternPg(False))
    initial = {"run_id": "v1_run_t", "policy_run_id": "policy_run_t", "platform": "kuaishou"}
    blocked = svc._gate1_blocked_state(initial, "gate1_no_terminal_element")
    assert blocked["status"] == "blocked"
    assert blocked["current_step"] == "blocked_gate1"
    assert blocked["error_code"] == "gate1_no_terminal_element"