fix: harden mirror material rebuild (empty/invalid frames, dedup, observability)
This commit is contained in:
@@ -2,7 +2,20 @@
|
||||
|
||||
import pytest
|
||||
|
||||
from app.question_gen.adversarial_filter import generate_mirror_question
|
||||
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
|
||||
|
||||
|
||||
@@ -73,3 +86,90 @@ async def test_mirror_malformed_response_returns_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"]
|
||||
|
||||
Reference in New Issue
Block a user