176 lines
5.7 KiB
Python
176 lines
5.7 KiB
Python
"""镜像生成：成功造出正解相反的镜像；正解相同/生成 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"]
|