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

"""M9E:Gate 2 query 级 50+ 判定——yes 保留/no 丢弃/拿不准从宽;judge 解析;游走标签复用。"""

import pytest

from content_agent.business_modules import search_intent
from content_agent.business_modules.search_intent import _gate2_keep
from content_agent.integrations import query_variant
from content_agent.integrations.query_variant import OpenRouterQueryVariantClient


class _Judge:
    def __init__(self, verdict):
        self.verdict = verdict
        self.calls: list[str] = []

    def judge_query_fifty_plus(self, query_text):
        self.calls.append(query_text)
        return self.verdict


class _RaisingJudge:
    def judge_query_fifty_plus(self, query_text):
        raise RuntimeError("llm down")


class _NoJudge:
    pass


def test_gate2_keeps_yes_and_drops_no():
    assert _gate2_keep("广场舞", _Judge(True)) is True
    assert _gate2_keep("二次元", _Judge(False)) is False


def test_gate2_keeps_on_error_and_missing_method():
    assert _gate2_keep("x", _RaisingJudge()) is True   # 拿不准从宽放过
    assert _gate2_keep("x", _NoJudge()) is True         # mock client 无 judge → 放过


def test_judge_parses_no_as_drop_and_else_as_keep(monkeypatch):
    client = OpenRouterQueryVariantClient(api_key="k", model="m")

    def fake_post(url, headers, json, timeout):
        import httpx

        answer = {"choices": [{"message": {"content": fake_post.reply}}]}
        return httpx.Response(200, json=answer, request=httpx.Request("POST", url))

    monkeypatch.setattr(query_variant.httpx, "post", fake_post)

    fake_post.reply = "no"
    assert client.judge_query_fifty_plus("二次元") is False
    fake_post.reply = "yes"
    assert client.judge_query_fifty_plus("广场舞") is True
    fake_post.reply = "不确定"
    assert client.judge_query_fifty_plus("模糊") is True  # 非 no 一律放行


def test_judge_keeps_on_http_exception(monkeypatch):
    client = OpenRouterQueryVariantClient(api_key="k", model="m")

    def boom(*args, **kwargs):
        raise RuntimeError("network")

    monkeypatch.setattr(query_variant.httpx, "post", boom)
    assert client.judge_query_fifty_plus("x") is True


class _Runtime:
    def __init__(self):
        self.rows: dict = {}

    def append_jsonl(self, run_id, filename, rows):
        self.rows.setdefault(filename, []).extend(rows)


def _qsp_seed_pack():
    return {
        "seed_terms": ["养生"],
        "query_seed_points": [
            {"point_text": "广场舞教学", "point_type": "灵感点", "rank": 1, "post_id": "p1", "id": "a1"},
            {"point_text": "二次元手办", "point_type": "灵感点", "rank": 2, "post_id": "p2", "id": "a2"},
            {"point_text": "钓鱼技巧", "point_type": "灵感点", "rank": 3, "post_id": "p3", "id": "a3"},
            {"point_text": "电竞比赛", "point_type": "灵感点", "rank": 4, "post_id": "p4", "id": "a4"},
        ],
        "pattern_execution_id": 1,
        "mining_config_id": 1,
        "source_post_id": "p1",
        "matched_post_ids": ["p1"],
        "itemset_ids": [],
    }


def test_gate2_rejects_query_seed_points_but_keeps_mandatory_seed_terms_for_non_douyin():
    # seed_terms 是首轮必搜;非抖音 Gate2 只过滤 query_seed_points。
    queries = search_intent.run(
        "r", "p", _qsp_seed_pack(), _Runtime(), _Judge(False),
        strategy_version="V4", platform="kuaishou",
    )
    assert [q["search_query"] for q in queries] == ["养生"]
    assert queries[0]["search_query_generation_method"] == "seed_term"
    assert "gate2_fallback" not in queries[0]["query_source_refs"][0]["source_ref"]


def test_douyin_skips_gate2_and_fallback():
    # M13:抖音豁免 Gate2,全产出、无 fallback 标记(零回归)。
    queries = search_intent.run(
        "r", "p", _qsp_seed_pack(), _Runtime(), _Judge(False),
        strategy_version="V4", platform="douyin",
    )
    assert len(queries) == 5
    assert all(
        "gate2_fallback" not in q["query_source_refs"][0]["source_ref"] for q in queries
    )