| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465 |
- """M9E:Gate 2 query 级 50+ 判定——yes 保留/no 丢弃/拿不准从宽;judge 解析;游走标签复用。"""
- import pytest
- 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
|