"""翻转门四路径：passed / filtered_no_flip / flip_skipped / 镜像不入库。""" import json import pytest from app.question_gen.adversarial_config import AdversarialFilterConfig from app.question_gen.adversarial_filter import ( agent_config_fingerprint, question_hash, run_flip_gate, write_final_bank, ) from app.question_gen.run_store import QuestionGenStore from core.types import GeneratedQuestion, LLMResponse class _FakeAgent: """复用 Task 6 语义；带 skill_mode 属性（AgentRunner Protocol 要求）。""" def __init__(self, preds, model="m1", skill_mode="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} class _FakeVLM: def __init__(self, content): self._content = content async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None): return LLMResponse( content=self._content, thinking="", model="fake", provider="fake", prompt_tokens=0, completion_tokens=0, latency_ms=0, ttft_ms=None, max_inter_token_ms=None, cache_hit=False, call_id="c", ) class _FakeMaterial: subtitle_sentences = ["先炒后蒸"] frame_paths = ["/f1.jpg"] def _q(qid, sub="temporal_reasoning_failure"): return GeneratedQuestion( question_id=qid, video_id="v1", task_type="Action Recognition", question="X 之前做了什么？", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A", source_nodes=("n1",), difficulty="hard", sub_pattern=sub, ) def _fp(): return agent_config_fingerprint(skill_mode="auto", max_steps=40, model="m1") @pytest.fixture(autouse=True) def _stub_material(monkeypatch): """隔离建树素材重建，直接给镜像生成喂假素材。""" monkeypatch.setattr( "app.question_gen.adversarial_filter._rebuild_material", lambda tree, source_nodes: _FakeMaterial(), ) def _preset_cheat(store, q, pred="A"): """预置作弊门 P 预测行（翻转门须复用它，不重跑 agent）。""" store.record_verdict( question_id=q.question_id, question_hash=question_hash(q), stage="cheat", round=0, agent_prediction=pred, agent_correct=False, verdict="passed", pair_id=None, agent_config=_fp(), ) @pytest.mark.asyncio async def test_flip_gate_answer_flips_passed(tmp_path): store = QuestionGenStore(str(tmp_path / "q.db")) q = _q("hard") _preset_cheat(store, q, pred="A") # P canonical="蒸" agent = _FakeAgent({"hard_mirror": "A"}) # 镜像洗牌后 A=炒 → canonical≠蒸 vlm = _FakeVLM( json.dumps( { "mirror": { "question": "X 之后？", "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A", } }, ensure_ascii=False, ) ) kept = await run_flip_gate( [q], agent=agent, vlm=vlm, store=store, trees={"v1": object()}, config=AdversarialFilterConfig(), round_no=0, run_id="r0", session_id="s", ) assert {x.question_id for x in kept} == {"hard"} assert agent.calls == ["hard_mirror"] # C2: 原题 P 未被重跑，只跑镜像 cheat = store._conn.execute( "SELECT verdict FROM adversarial_verdicts WHERE question_id='hard' AND stage='cheat'" ).fetchone()[0] assert cheat == "passed" mrow = store._conn.execute( "SELECT verdict, pair_id FROM adversarial_verdicts WHERE stage='flip_mirror'" ).fetchone() assert mrow[0] == "passed" and mrow[1] # 镜像独立行 + pair_id 非空 store.close() @pytest.mark.asyncio async def test_flip_gate_same_answer_filtered(tmp_path): store = QuestionGenStore(str(tmp_path / "q.db")) q = _q("stick") _preset_cheat(store, q, pred="A") # P canonical="蒸" agent = _FakeAgent({"stick_mirror": "A"}) # 镜像 A=蒸 → canonical 与 P 相同 vlm = _FakeVLM( json.dumps( { "mirror": { "question": "X 之后？", "options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], "answer": "B", } }, ensure_ascii=False, ) ) kept = await run_flip_gate( [q], agent=agent, vlm=vlm, store=store, trees={"v1": object()}, config=AdversarialFilterConfig(), round_no=0, run_id="r0", session_id="s", ) assert kept == [] # 未随问题翻转 → 剔除 cheat = store._conn.execute( "SELECT verdict FROM adversarial_verdicts WHERE question_id='stick' AND stage='cheat'" ).fetchone()[0] assert cheat == "filtered_no_flip" # cheat 行被改写 → final 不含它 store.close() @pytest.mark.asyncio async def test_flip_gate_invalid_mirror_skipped_but_kept(tmp_path): store = QuestionGenStore(str(tmp_path / "q.db")) q = _q("murky") _preset_cheat(store, q, pred="A") agent = _FakeAgent({}) # 镜像造不出 → agent 不该被调用 vlm = _FakeVLM('{"mirror": null}') kept = await run_flip_gate( [q], agent=agent, vlm=vlm, store=store, trees={"v1": object()}, config=AdversarialFilterConfig(), round_no=0, run_id="r0", session_id="s", ) assert {x.question_id for x in kept} == {"murky"} # 退回只经作弊门，保留不误杀 assert agent.calls == [] # 镜像 None → 未跑 agent cheat = store._conn.execute( "SELECT verdict FROM adversarial_verdicts WHERE question_id='murky' AND stage='cheat'" ).fetchone()[0] assert cheat == "passed" store.close() @pytest.mark.asyncio async def test_flip_gate_mirror_excluded_and_unsupported_passes(tmp_path): store = QuestionGenStore(str(tmp_path / "q.db")) q = _q("hard") # 支持 flip npq = _q("plain", sub="premature_evidence_anchoring") # 不支持 flip _preset_cheat(store, q, pred="A") _preset_cheat(store, npq, pred="B") agent = _FakeAgent({"hard_mirror": "A"}) vlm = _FakeVLM( json.dumps( { "mirror": { "question": "X 之后？", "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A", } }, ensure_ascii=False, ) ) kept = await run_flip_gate( [q, npq], agent=agent, vlm=vlm, store=store, trees={"v1": object()}, config=AdversarialFilterConfig(), round_no=0, run_id="r0", session_id="s", ) assert {x.question_id for x in kept} == {"hard", "plain"} # 不支持 flip 直接 passed assert agent.calls == ["hard_mirror"] # 不支持 flip 的题不跑 agent/VLM out = tmp_path / "final.json" write_final_bank(out, store, {"hard": q, "plain": npq}, _fp()) ids = [d["question_id"] for d in json.loads(out.read_text(encoding="utf-8"))] assert "hard_mirror" not in ids and set(ids) == {"hard", "plain"} # 镜像不入题库 store.close()