iomgaa
f57ee45dc0
Critical: - C1: assemble_mode 'plain' → 'ids'（合法枚举值） - C2: question_id 加入 task_type slug 避免跨题型冲突 Important/Minor: - generate_one 移除未用的 embed_fn/similarity_threshold 参数 - config.py 注释 11→12 同步 - 测试 question_id 断言更新 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
491 lines
19 KiB
Python
491 lines
19 KiB
Python
"""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-object_recognition-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-object_recognition-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-action_reasoning-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"}',
|
||
)
|
||
|
||
tree, vid = self._load_test_tree()
|
||
result = await generate_one(
|
||
vlm=vlm,
|
||
tree=tree,
|
||
video_id=vid,
|
||
task_type="Object Recognition",
|
||
seq=1,
|
||
exemplars=[],
|
||
used_node_ids=set(),
|
||
max_retries=3,
|
||
rng=random.Random(42),
|
||
session_id="test",
|
||
)
|
||
assert result is not None
|
||
assert result.question_id == f"gen-{vid}-object_recognition-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")
|
||
|
||
tree, vid = self._load_test_tree()
|
||
result = await generate_one(
|
||
vlm=vlm,
|
||
tree=tree,
|
||
video_id=vid,
|
||
task_type="Object Recognition",
|
||
seq=1,
|
||
exemplars=[],
|
||
used_node_ids=set(),
|
||
max_retries=2,
|
||
rng=random.Random(42),
|
||
session_id="test",
|
||
)
|
||
assert result is None
|
||
assert vlm.chat_with_images.call_count == 2
|