"""镜像生成:成功造出正解相反的镜像;正解相同/生成 null → 返回 None。""" import pytest from app.question_gen.adversarial_filter import ( _rebuild_material, generate_mirror_question, ) from app.tree.index import ( IndexMeta, L1Card, L1Node, L2Card, L2Node, L3Card, L3Node, TreeIndex, ) from core.types import GeneratedQuestion, LLMResponse class _FakeVLM: def __init__(self, content: str): 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", ) def _q(): return GeneratedQuestion( question_id="q1", video_id="v1", task_type="Action Recognition", question="X 之前做了什么?", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A", source_nodes=("n1",), difficulty="hard", sub_pattern="temporal_reasoning_failure", ) @pytest.mark.asyncio async def test_mirror_distinct_correct_ok(): vlm = _FakeVLM('{"mirror": {"question": "X 之后做了什么?", ' '"options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}') mirror = await generate_mirror_question( _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", ) assert mirror is not None # 原正解 canonical="蒸",镜像正解 canonical="炒" → 相异,有效 assert mirror.answer == "A" assert mirror.options[0] == "A. 炒" @pytest.mark.asyncio async def test_mirror_same_correct_rejected(): # 镜像正解 canonical 仍是"蒸" → 造不出有效对 → None vlm = _FakeVLM('{"mirror": {"question": "X 之后?", ' '"options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], "answer": "A"}}') mirror = await generate_mirror_question( _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", ) assert mirror is None @pytest.mark.asyncio async def test_mirror_null_returns_none(): vlm = _FakeVLM('{"mirror": null}') mirror = await generate_mirror_question( _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", ) assert mirror is None @pytest.mark.asyncio async def test_mirror_malformed_response_returns_none(): # 畸形 VLM 响应(连 json_repair 都救不回)不得抛异常中断本轮,须返 None vlm = _FakeVLM("对不起,我无法完成这个请求。") mirror = await generate_mirror_question( _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", ) assert mirror is None class _FakeMaterial: subtitle_sentences = ["先炒后蒸"] frame_paths = ["/f1.jpg"] class _NoCallVLM: """素材为空时被误调用即失败——断言空素材绝不触达 VLM。""" async def chat_with_images(self, *args, **kwargs): raise AssertionError("素材为空时不应调用 VLM") class _EmptyMaterial: subtitle_sentences: list[str] = [] frame_paths = ["/does/not/exist/frame_x.jpg"] @pytest.mark.asyncio async def test_mirror_empty_material_skips_vlm(): # 无字幕 AND 帧全不存在 → 直接返 None,且绝不调用 VLM mirror = await generate_mirror_question( _q(), flip_axis="before/after", vlm=_NoCallVLM(), material=_EmptyMaterial(), session_id="s", ) assert mirror is None def _tiny_tree() -> tuple[TreeIndex, str, str]: """构建 1×L1→1×L2→2×L3 的最小真实树;返回 (tree, l2_id, first_l3_id)。 L2 自带字幕,两个 L3 各带字幕与帧路径,便于验证父子重叠去重。 """ l3_nodes = [ L3Node( id=f"l2a_l3_{i}", card=L3Card( frame_summary=f"帧{i}", visible_entities=[], ongoing_actions=[], visible_text=[], spatial_layout="居中", visual_attributes={}, subtitle=f"字幕{i}", ), frame_path=f"frames/l2a_l3_{i}.jpg", ) for i in range(2) ] l2 = L2Node( id="l2a", card=L2Card( event_description="事件A", entities=[], actions=[], action_subjects=[], visible_text=[], spatial_relations="并列", state_changes=None, subtitle="L2字幕", ), children=l3_nodes, ) l1 = L1Node( id="l1a", card=L1Card( scene_summary="场景A", main_setting="室内", key_entities=[], main_actions=[], topic_keywords=[], visible_text=[], temporal_flow="顺序", ), children=[l2], ) tree = TreeIndex( metadata=IndexMeta(source_path="/t.mp4", modality="video"), roots=[l1] ) return tree, "l2a", "l2a_l3_0" def test_rebuild_material_dedup_parent_child(): # source_nodes 同含父 L2 与子 L3(子树重叠)→ 素材须保序去重、无冗余 tree, l2_id, l3_id = _tiny_tree() material = _rebuild_material(tree, (l2_id, l3_id)) assert len(material.subtitle_sentences) == len(set(material.subtitle_sentences)) assert len(material.frame_paths) == len(set(material.frame_paths)) # L2字幕 + 两条 L3 字幕 = 3 条;两帧 = 2 帧(去重后) assert material.subtitle_sentences == ["L2字幕", "字幕0", "字幕1"] assert material.frame_paths == ["frames/l2a_l3_0.jpg", "frames/l2a_l3_1.jpg"]