"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor + prompt 构造 + VLM 解析 + 去重 + 单题生成。""" from __future__ import annotations import dataclasses import random from pathlib import Path from unittest.mock import AsyncMock, MagicMock import numpy as np import pytest from app.question_gen.synthesizer import ( TASK_TYPE_LEVEL_MAP, AnchorContext, TaskTypeSpec, build_generation_prompt, generate_one, is_duplicate, parse_vlm_response, sample_anchor, ) from app.tree.index import TreeIndex from core.types import GeneratedQuestion ALL_12_TYPES = [ "Object Recognition", "Attribute Perception", "OCR Problems", "Spatial Reasoning", "Spatial Perception", "Action Recognition", "Action Reasoning", "Counting Problem", "Temporal Perception", "Temporal Reasoning", "Information Synopsis", "Object Reasoning", ] class TestTaskTypeLevelMap: """TASK_TYPE_LEVEL_MAP 覆盖性与结构测试。""" def test_covers_all_12_types(self) -> None: """映射表必须覆盖全部 12 种 Video-MME 题型。""" assert set(TASK_TYPE_LEVEL_MAP.keys()) == set(ALL_12_TYPES) def test_no_extra_types(self) -> None: """映射表不得包含 12 种标准题型之外的条目。""" assert len(TASK_TYPE_LEVEL_MAP) == 12 def test_all_values_are_task_type_spec(self) -> None: """每个映射值必须是 TaskTypeSpec 实例。""" for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): assert isinstance(spec, TaskTypeSpec), f"{task_type} 映射值类型错误: {type(spec)}" def test_level_values_valid(self) -> None: """每个 spec 的 level 必须是合法层级标识。""" valid_levels = {"L1", "L2", "L3", "L1-L2"} for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): assert spec.level in valid_levels, ( f"{task_type} 层级 '{spec.level}' 不在 {valid_levels}" ) def test_context_fields_non_empty(self) -> None: """每个 spec 的 context_fields 至少有一个字段。""" for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): assert len(spec.context_fields) >= 1, f"{task_type} 的 context_fields 为空" class TestAnchorContext: """AnchorContext 数据类测试。""" def test_frozen(self) -> None: """AnchorContext 是不可变的。""" ctx = AnchorContext( node_id="L3_001", card_text="A person walks into a room", frame_paths=["/data/frames/001.jpg"], subtitle="Hello there", distractor_texts=["A car drives by"], ) assert ctx.node_id == "L3_001" assert ctx.card_text == "A person walks into a room" assert ctx.frame_paths == ["/data/frames/001.jpg"] assert ctx.subtitle == "Hello there" assert ctx.distractor_texts == ["A car drives by"] def test_mutation_raises(self) -> None: """frozen dataclass 拒绝赋值修改。""" ctx = AnchorContext( node_id="L3_001", card_text="test", frame_paths=["a.jpg"], subtitle="", distractor_texts=["other node"], ) try: ctx.node_id = "L3_002" # type: ignore[misc] raise AssertionError("应抛出 FrozenInstanceError") except dataclasses.FrozenInstanceError: pass def test_empty_subtitle_allowed(self) -> None: """subtitle 可以为空字符串。""" ctx = AnchorContext( node_id="L2_010", card_text="scene card", frame_paths=[], subtitle="", distractor_texts=[], ) assert ctx.subtitle == "" def test_multiple_frame_paths(self) -> None: """frame_paths 可包含多个路径。""" paths = ["/data/f1.jpg", "/data/f2.jpg", "/data/f3.jpg"] ctx = AnchorContext( node_id="L2_005", card_text="multi-frame event", frame_paths=paths, subtitle="Dialogue line", distractor_texts=["other1", "other2"], ) assert len(ctx.frame_paths) == 3 class TestTaskTypeSpec: """TaskTypeSpec 数据类测试。""" def test_frozen(self) -> None: """TaskTypeSpec 是不可变的。""" spec = TaskTypeSpec( level="L3", needs_frames=True, frame_count="1", context_fields=("frame_summary",), ) try: spec.level = "L2" # type: ignore[misc] raise AssertionError("应抛出 FrozenInstanceError") except dataclasses.FrozenInstanceError: pass def test_context_fields_is_tuple(self) -> None: """context_fields 应为 tuple（不可变）。""" for task_type, spec in TASK_TYPE_LEVEL_MAP.items(): assert isinstance(spec.context_fields, tuple), ( f"{task_type} 的 context_fields 不是 tuple" ) # --------------------------------------------------------------------------- # sample_anchor 测试 # --------------------------------------------------------------------------- def _load_test_tree() -> tuple[TreeIndex, str]: """加载真实测试树（store/videos/ 下第一棵）。""" videos_dir = Path("store/videos") first_vid = sorted(videos_dir.iterdir())[0] tree = TreeIndex.load_json(str(first_vid / "tree.json")) return tree, first_vid.name class TestSampleAnchor: """sample_anchor 锚节点采样测试（基于真实树数据）。""" def test_l3_type_returns_single_frame(self) -> None: """L3 题型（Object Recognition）应返回恰好 1 帧。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Object Recognition", set(), random.Random(42)) assert len(ctx.frame_paths) == 1 assert ctx.node_id.startswith("L") or "_L3_" in ctx.node_id assert len(ctx.distractor_texts) > 0 def test_l2_type_returns_multiple_frames(self) -> None: """L2 题型（Action Reasoning）应返回 2-3 帧。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Action Reasoning", set(), random.Random(42)) assert 2 <= len(ctx.frame_paths) <= 3 def test_temporal_perception_zero_or_one_frame(self) -> None: """Temporal Perception 特殊处理：0-1 帧，且 card_text 包含 time_range。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Temporal Perception", set(), random.Random(42)) assert len(ctx.frame_paths) <= 1 assert "time_range" in ctx.card_text.lower() or "time" in ctx.card_text.lower() def test_information_synopsis_uses_all_l2(self) -> None: """Information Synopsis 必须使用目标 L1 下所有 L2 子节点。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Information Synopsis", set(), random.Random(42)) # 找到被选中的 L1，验证 frame_paths 数量 == 该 L1 下全部 L2 数量 chosen_l1 = next(r for r in tree.roots if r.id == ctx.node_id) assert len(ctx.frame_paths) == len(chosen_l1.children) def test_l1_type_l2_nodes_in_time_order(self) -> None: """Temporal Reasoning 应返回 >=3 帧且 card_text 有实质内容。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Temporal Reasoning", set(), random.Random(42)) assert len(ctx.frame_paths) >= 3 assert len(ctx.card_text) > 20 def test_used_node_ids_excluded(self) -> None: """used_node_ids 中的节点不应被再次选中。""" tree, _vid = _load_test_tree() rng = random.Random(42) ctx1 = sample_anchor(tree, "Object Recognition", set(), rng) ctx2 = sample_anchor(tree, "Object Recognition", {ctx1.node_id}, random.Random(43)) assert ctx2.node_id != ctx1.node_id def test_insufficient_nodes_raises(self) -> None: """所有候选节点均被排除时应抛出 ValueError。""" tree, _vid = _load_test_tree() all_l3_ids: set[str] = set() for root in tree.roots: for l2 in root.children: for l3 in l2.children: all_l3_ids.add(l3.id) with pytest.raises(ValueError, match="锚节点不足"): sample_anchor(tree, "Object Recognition", all_l3_ids, random.Random(42)) def test_object_reasoning_l1_l2_type(self) -> None: """Object Reasoning (L1-L2) 应选 2-3 个 L2 并按时间排序。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Object Reasoning", set(), random.Random(42)) assert 2 <= len(ctx.frame_paths) <= 3 assert len(ctx.card_text) > 10 def test_spatial_reasoning_context_fields(self) -> None: """Spatial Reasoning 的 card_text 应包含 spatial_layout 字段。""" tree, _vid = _load_test_tree() ctx = sample_anchor(tree, "Spatial Reasoning", set(), random.Random(42)) # context_fields 包含 spatial_layout，card_text 必须出现该字段名 assert "spatial_layout" in ctx.card_text assert len(ctx.card_text) > 20 # --------------------------------------------------------------------------- # build_generation_prompt 测试 # --------------------------------------------------------------------------- class TestBuildGenerationPrompt: """build_generation_prompt 消息结构与内容测试。""" def test_messages_structure(self) -> None: """带 exemplars 时，system 包含题型和示例，image_paths 来自 anchor。""" anchor = AnchorContext( node_id="L3_001", card_text="A person typing on a laptop", frame_paths=["store/videos/test/frames/L1_000_L2_000_L3_000.jpg"], subtitle="Hello world", distractor_texts=["Another person walking in park"], ) exemplars = [ GeneratedQuestion( question_id="ex-1", video_id="v1", task_type="Object Recognition", question="What object?", options=("A. Cat", "B. Dog", "C. Bird", "D. Fish"), answer="A", source_nodes=(), difficulty="medium", ), ] messages, image_paths = build_generation_prompt( "Object Recognition", anchor, exemplars, ) assert messages[0]["role"] == "system" assert "Object Recognition" in messages[0]["content"] assert any("What object?" in str(m) for m in messages) assert image_paths == anchor.frame_paths def test_distractor_in_user_message(self) -> None: """干扰项文本应出现在 user message 中。""" anchor = AnchorContext( node_id="L2_003", card_text="Event card text", frame_paths=["a.jpg", "b.jpg"], subtitle="", distractor_texts=["Distractor node summary"], ) messages, _ = build_generation_prompt("Action Reasoning", anchor, []) user_msg = [m for m in messages if m["role"] == "user"][0] assert "Distractor node summary" in user_msg["content"] def test_no_exemplars_no_crash(self) -> None: """exemplars 为空时不应报错，system 消息中无示例段落。""" anchor = AnchorContext( node_id="L3_010", card_text="Some card text", frame_paths=["frame.jpg"], subtitle="", distractor_texts=[], ) messages, image_paths = build_generation_prompt("OCR Problems", anchor, []) assert len(messages) >= 2 assert image_paths == ["frame.jpg"] def test_subtitle_included_when_non_empty(self) -> None: """非空 subtitle 应出现在 user message 中。""" anchor = AnchorContext( node_id="L3_002", card_text="Card text here", frame_paths=["f.jpg"], subtitle="This is a subtitle line", distractor_texts=[], ) messages, _ = build_generation_prompt("Attribute Perception", anchor, []) user_msg = [m for m in messages if m["role"] == "user"][0] assert "This is a subtitle line" in user_msg["content"] def test_empty_subtitle_not_in_user_message(self) -> None: """空 subtitle 不应在 user message 中产生 subtitle 段落。""" anchor = AnchorContext( node_id="L3_003", card_text="Card", frame_paths=["f.jpg"], subtitle="", distractor_texts=[], ) messages, _ = build_generation_prompt("OCR Problems", anchor, []) user_msg = [m for m in messages if m["role"] == "user"][0] # 不应出现空的 subtitle 标记 assert "字幕" not in user_msg["content"] # --------------------------------------------------------------------------- # parse_vlm_response 测试 # --------------------------------------------------------------------------- class TestParseVlmResponse: """parse_vlm_response 解析与校验测试。""" def test_valid_json(self) -> None: """合法 JSON 正常解析，question_id 格式正确。""" raw = '{"question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A"}' result = parse_vlm_response(raw, "vid1", "Object Recognition", 1) assert result["question"] == "What?" assert result["answer"] == "A" assert len(result["options"]) == 4 assert result["question_id"] == "gen-vid1-001" def test_json_in_code_block(self) -> None: """从 markdown 代码块中提取 JSON。""" raw = '```json\n{"question": "Q?", "options": ["A. 1", "B. 2", "C. 3", "D. 4"], "answer": "B"}\n```' result = parse_vlm_response(raw, "vid1", "Object Recognition", 2) assert result["question"] == "Q?" assert result["question_id"] == "gen-vid1-002" def test_invalid_json_raises(self) -> None: """非 JSON 文本应抛出 ValueError。""" with pytest.raises(ValueError, match="VLM 返回"): parse_vlm_response("not json", "vid1", "Object Recognition", 1) def test_missing_fields_raises(self) -> None: """缺少必需字段应抛出 ValueError。""" raw = '{"question": "What?"}' with pytest.raises(ValueError): parse_vlm_response(raw, "vid1", "Object Recognition", 1) def test_options_must_be_four(self) -> None: """options 非 4 项应抛出 ValueError。""" raw = '{"question": "Q?", "options": ["A. X", "B. Y"], "answer": "A"}' with pytest.raises(ValueError, match="4"): parse_vlm_response(raw, "vid1", "Object Recognition", 1) def test_answer_must_be_abcd(self) -> None: """answer 不在 A-D 范围应抛出 ValueError。""" raw = '{"question": "Q?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "E"}' with pytest.raises(ValueError, match="A.*D"): parse_vlm_response(raw, "vid1", "Object Recognition", 1) def test_seq_zero_padded(self) -> None: """seq 应按 3 位零填充格式化到 question_id 中。""" raw = '{"question": "Q?", "options": ["A. 1", "B. 2", "C. 3", "D. 4"], "answer": "C"}' result = parse_vlm_response(raw, "video_abc", "Action Reasoning", 42) assert result["question_id"] == "gen-video_abc-042" # --------------------------------------------------------------------------- # is_duplicate 测试 # --------------------------------------------------------------------------- class TestIsDuplicate: """is_duplicate embedding 去重判定测试。""" @staticmethod def _fake_embed(texts: str | list[str]) -> np.ndarray: """确定性 + L2 归一化的 fake embedding。""" if isinstance(texts, str): texts = [texts] vecs = [] for t in texts: rs = np.random.RandomState(hash(t) % 2**31) v = rs.randn(4).astype(np.float32) v /= np.linalg.norm(v) vecs.append(v) return np.array(vecs, dtype=np.float32) def test_empty_pool_never_duplicate(self) -> None: """空池始终返回 False。""" pool = np.zeros((0, 4), dtype=np.float32) assert is_duplicate("anything", pool, self._fake_embed, 0.85) is False def test_identical_text_is_duplicate(self) -> None: """相同文本的 embedding 与自身余弦相似度为 1，必定判重。""" text = "What is happening in the video?" emb = self._fake_embed(text) pool = emb.copy() assert is_duplicate(text, pool, self._fake_embed, 0.85) is True def test_different_text_not_duplicate(self) -> None: """极高阈值下，不同文本不判重。""" pool_texts = ["aaa", "bbb", "ccc", "ddd", "eee"] pool = self._fake_embed(pool_texts) assert is_duplicate("completely unique text xyz", pool, self._fake_embed, 0.99) is False # --------------------------------------------------------------------------- # generate_one 测试 # --------------------------------------------------------------------------- class TestGenerateOne: """generate_one 单题异步生成测试。""" @staticmethod def _load_test_tree() -> tuple[TreeIndex, str]: """加载真实测试树。""" videos_dir = Path("store/videos") first_vid = sorted(videos_dir.iterdir())[0] return TreeIndex.load_json(str(first_vid / "tree.json")), first_vid.name @pytest.mark.asyncio async def test_success_path(self) -> None: """mock VLM 返回合法 JSON，应成功生成 GeneratedQuestion。""" vlm = AsyncMock() vlm.chat_with_images.return_value = MagicMock( content='{"question":"Q?","options":["A. 1","B. 2","C. 3","D. 4"],"answer":"A"}', ) def embed_fn(t: str | list[str]) -> np.ndarray: shape = (1, 4) if isinstance(t, str) else (len(t), 4) return np.zeros(shape, dtype=np.float32) tree, vid = self._load_test_tree() result = await generate_one( vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid, task_type="Object Recognition", seq=1, exemplars=[], used_node_ids=set(), max_retries=3, similarity_threshold=0.85, rng=random.Random(42), session_id="test", ) assert result is not None assert result.question_id == f"gen-{vid}-001" assert result.task_type == "Object Recognition" assert result.source_nodes # non-empty assert result.difficulty == "medium" @pytest.mark.asyncio async def test_all_retries_exhausted_returns_none(self) -> None: """VLM 始终返回无效 JSON，耗尽重试后返回 None。""" vlm = AsyncMock() vlm.chat_with_images.return_value = MagicMock(content="invalid") def embed_fn(t: str | list[str]) -> np.ndarray: return np.zeros((1, 4), dtype=np.float32) tree, vid = self._load_test_tree() result = await generate_one( vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid, task_type="Object Recognition", seq=1, exemplars=[], used_node_ids=set(), max_retries=2, similarity_threshold=0.85, rng=random.Random(42), session_id="test", ) assert result is None assert vlm.chat_with_images.call_count == 2