88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
"""作弊门：agent 答对=filtered_too_easy 并落表；答错=cheat verdict=passed 待翻转。"""
|
||
|
||
import pytest
|
||
|
||
from app.question_gen.adversarial_config import AdversarialFilterConfig
|
||
from app.question_gen.adversarial_filter import run_cheater_gate
|
||
from app.question_gen.run_store import QuestionGenStore
|
||
from core.types import GeneratedQuestion
|
||
|
||
|
||
class _FakeAgent:
|
||
"""按 question_id → 预测字母返回的 mock AgentRunner。"""
|
||
|
||
def __init__(self, preds: dict[str, str], model: str = "m1", skill_mode: str = "auto"):
|
||
self._preds = preds
|
||
self.model = model
|
||
self.skill_mode = skill_mode
|
||
self.calls: list[str] = []
|
||
|
||
async def predict(self, questions, *, max_steps, run_id):
|
||
self.calls.extend(q.question_id for q in questions)
|
||
return {q.question_id: self._preds.get(q.question_id) for q in questions}
|
||
|
||
|
||
def _q(qid, answer="A"):
|
||
return GeneratedQuestion(
|
||
question_id=qid, video_id="v1", task_type="Action Recognition",
|
||
question="?", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer=answer,
|
||
source_nodes=("n1",), difficulty="hard", sub_pattern="temporal_reasoning_failure",
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cheater_gate_filters_too_easy_and_keeps_hard(tmp_path):
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
agent = _FakeAgent({"easy": "A", "hard": "B"}) # easy 答对(A), hard 答错
|
||
cfg = AdversarialFilterConfig()
|
||
survivors = await run_cheater_gate(
|
||
[_q("easy"), _q("hard")], agent=agent, store=store,
|
||
config=cfg, round_no=0, run_id="r0",
|
||
)
|
||
ids = {q.question_id for q in survivors}
|
||
assert ids == {"hard"} # 只有答错的进翻转门
|
||
verdicts = {
|
||
r[0]: r[1] for r in store._conn.execute(
|
||
"SELECT question_id, verdict FROM adversarial_verdicts WHERE stage='cheat'"
|
||
)
|
||
}
|
||
assert verdicts["easy"] == "filtered_too_easy"
|
||
# hard 在 cheat 阶段先记 passed（待翻转门可能改写；不支持翻转的题即终判 passed）
|
||
assert verdicts["hard"] == "passed"
|
||
store.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cheater_gate_resume_skips_completed(tmp_path):
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
agent = _FakeAgent({"hard": "B"})
|
||
cfg = AdversarialFilterConfig()
|
||
await run_cheater_gate([_q("hard")], agent=agent, store=store,
|
||
config=cfg, round_no=0, run_id="r0")
|
||
first = list(agent.calls)
|
||
await run_cheater_gate([_q("hard")], agent=agent, store=store,
|
||
config=cfg, round_no=0, run_id="r1")
|
||
assert agent.calls == first # 第二次不重跑（已有 cheat verdict）
|
||
store.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cheater_gate_resume_mixed_keeps_all_survivors(tmp_path):
|
||
"""混合续跑：部分题已有 cheat verdict、部分未判——已完成的存活者不得被丢。"""
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
cfg = AdversarialFilterConfig()
|
||
# 第一轮：先只判 done_hard（答错=存活），落表
|
||
agent1 = _FakeAgent({"done_hard": "B"}) # 答错(正解 A)
|
||
await run_cheater_gate([_q("done_hard")], agent=agent1, store=store,
|
||
config=cfg, round_no=0, run_id="r0")
|
||
# 第二轮：done_hard 已判 + 新题 new_hard 未判混在一起
|
||
agent2 = _FakeAgent({"new_hard": "C"}) # 新题答错(正解 A)=存活
|
||
survivors = await run_cheater_gate(
|
||
[_q("done_hard"), _q("new_hard")], agent=agent2, store=store,
|
||
config=cfg, round_no=0, run_id="r1",
|
||
)
|
||
ids = {q.question_id for q in survivors}
|
||
assert ids == {"done_hard", "new_hard"} # 已完成存活者 done_hard 未被丢
|
||
assert agent2.calls == ["new_hard"] # 只对未判题跑 agent
|
||
store.close()
|