Video-Tree-TRM5/tests/unit/test_sampler_v2.py

"""v2 素材采样器单元测试。

测试 sample_material_v2 及其辅助函数的核心行为：
- 正常采样返回 MaterialContext
- 已用节点排除
- 约束违反时重试直至 RuntimeError
- 跨 L2 上下文收集
- 字幕收集
"""

from __future__ import annotations

import random

import pytest

from app.question_gen.families import (
    REASONING_FAMILY,
    RETRIEVAL_FAMILY,
    VISUAL_FAMILY,
    SamplingConstraint,
)
from app.tree.index import (
    IndexMeta,
    L1Card,
    L1Node,
    L2Card,
    L2Node,
    L3Card,
    L3Node,
    TreeIndex,
)

# ---------------------------------------------------------------------------
# Fixture: 构建含丰富数据的真实树结构
# ---------------------------------------------------------------------------


def _make_l3(
    l1_idx: int,
    l2_idx: int,
    l3_idx: int,
    *,
    subtitle: str = "",
    frame_path: str | None = None,
) -> L3Node:
    """构建 L3 节点，带可控 subtitle/frame_path。"""
    node_id = f"l1_{l1_idx}_l2_{l2_idx}_l3_{l3_idx}"
    return L3Node(
        id=node_id,
        card=L3Card(
            frame_summary=f"帧{l3_idx}描述:L1={l1_idx},L2={l2_idx}",
            visible_entities=[f"实体_{l3_idx}"],
            ongoing_actions=[f"动作_{l3_idx}"],
            visible_text=[],
            spatial_layout="居中",
            visual_attributes={"lighting": "明亮"},
            subtitle=subtitle,
        ),
        timestamp=float(l3_idx * 2),
        frame_path=frame_path,
    )


def _make_l2(
    l1_idx: int,
    l2_idx: int,
    n_l3: int = 3,
    *,
    subtitle: str = "",
    with_frames: bool = True,
    with_subtitles: bool = True,
) -> L2Node:
    """构建 L2 节点，可控子节点数量和属性。"""
    children: list[L3Node] = []
    for i in range(n_l3):
        sub = f"字幕L1={l1_idx}_L2={l2_idx}_L3={i}" if with_subtitles else ""
        fp = f"frames/l1_{l1_idx}_l2_{l2_idx}_l3_{i}.jpg" if with_frames else None
        children.append(_make_l3(l1_idx, l2_idx, i, subtitle=sub, frame_path=fp))

    l2_subtitle = subtitle or (f"L2事件字幕:L1={l1_idx}_L2={l2_idx}" if with_subtitles else "")
    return L2Node(
        id=f"l1_{l1_idx}_l2_{l2_idx}",
        card=L2Card(
            event_description=f"事件:L1={l1_idx},L2={l2_idx}",
            entities=[f"角色_{l2_idx}"],
            actions=[f"行为_{l2_idx}"],
            action_subjects=[f"主体_{l2_idx}"],
            visible_text=[],
            spatial_relations="左右排列",
            state_changes=None,
            subtitle=l2_subtitle,
        ),
        time_range=(l2_idx * 30.0, (l2_idx + 1) * 30.0),
        children=children,
    )


def _make_l1(l1_idx: int, n_l2: int = 3, n_l3: int = 3) -> L1Node:
    """构建 L1 节点，含多个 L2 子节点。"""
    return L1Node(
        id=f"l1_{l1_idx}",
        card=L1Card(
            scene_summary=f"场景{l1_idx}摘要",
            main_setting="室内" if l1_idx % 2 == 0 else "户外",
            key_entities=[f"主角_{l1_idx}"],
            main_actions=[f"主行为_{l1_idx}"],
            topic_keywords=[f"关键词_{l1_idx}"],
            visible_text=[],
            temporal_flow="从左到右",
        ),
        time_range=(l1_idx * 600.0, (l1_idx + 1) * 600.0),
        children=[_make_l2(l1_idx, j, n_l3) for j in range(n_l2)],
    )


@pytest.fixture()
def real_tree() -> TreeIndex:
    """构建包含 2 个 L1、每个 L1 含 3 个 L2、每个 L2 含 5 个 L3 的真实树。

    共 2*3*5 = 30 个 L3 节点，6 个 L2 节点，2 个 L1 节点。
    所有节点有帧路径和字幕。满足 REASONING_FAMILY 的 min_l3_nodes=4 要求。
    """
    meta = IndexMeta(source_path="/test/video.mp4", modality="video")
    roots = [_make_l1(i, n_l2=3, n_l3=5) for i in range(2)]
    return TreeIndex(metadata=meta, roots=roots)


@pytest.fixture()
def sparse_tree() -> TreeIndex:
    """构建一棵稀疏树——无帧、少字幕，用于测试约束违反。

    只有 1 个 L1, 1 个 L2, 1 个 L3。L3 无帧无字幕。
    """
    meta = IndexMeta(source_path="/test/sparse.mp4", modality="video")
    l3 = _make_l3(0, 0, 0, subtitle="", frame_path=None)
    l2 = L2Node(
        id="sparse_l2_0",
        card=L2Card(
            event_description="稀疏事件",
            entities=[],
            actions=[],
            action_subjects=[],
            visible_text=[],
            spatial_relations="",
            state_changes=None,
            subtitle="",
        ),
        time_range=(0.0, 30.0),
        children=[l3],
    )
    l1 = L1Node(
        id="sparse_l1_0",
        card=L1Card(
            scene_summary="稀疏场景",
            main_setting="未知",
            key_entities=[],
            main_actions=[],
            topic_keywords=[],
            visible_text=[],
            temporal_flow="",
        ),
        time_range=(0.0, 600.0),
        children=[l2],
    )
    return TreeIndex(metadata=meta, roots=[l1])


# ---------------------------------------------------------------------------
# 测试类
# ---------------------------------------------------------------------------


class TestSampleMaterialV2:
    """sample_material_v2 核心行为测试。"""

    def test_returns_material_context(self, real_tree: TreeIndex) -> None:
        """正常采样返回 MaterialContext，字段类型正确。"""
        from app.question_gen.sampler_v2 import MaterialContext, sample_material_v2

        rng = random.Random(42)
        result = sample_material_v2(
            tree=real_tree,
            task_type="Action Reasoning",
            used_node_ids=set(),
            rng=rng,
            level=2,
            constraint=RETRIEVAL_FAMILY.sampling,
        )

        assert isinstance(result, MaterialContext)
        assert result.anchor.node_id  # 非空
        assert result.anchor.level in (1, 2, 3)
        assert len(result.source_nodes) > 0
        assert isinstance(result.subtitle_sentences, list)
        assert isinstance(result.frame_paths, list)
        assert isinstance(result.cross_l2_texts, list)

    def test_respects_used_nodes(self, real_tree: TreeIndex) -> None:
        """已用节点被正确排除，不会重复采样。"""
        from app.question_gen.sampler_v2 import sample_material_v2

        rng = random.Random(42)

        # 把所有 L2 节点标记为已用（除了最后一个）
        all_l2_ids: set[str] = set()
        for l1 in real_tree.roots:
            for l2 in l1.children:
                all_l2_ids.add(l2.id)

        # 留下恰好一个 L2 未用
        last_l2_id = real_tree.roots[-1].children[-1].id
        used = all_l2_ids - {last_l2_id}

        result = sample_material_v2(
            tree=real_tree,
            task_type="Action Reasoning",
            used_node_ids=used,
            rng=rng,
            level=2,
            constraint=RETRIEVAL_FAMILY.sampling,
        )

        # 锚节点应该是那个未被排除的 L2
        assert result.anchor.node_id == last_l2_id

    def test_constraint_violation_retries(self, sparse_tree: TreeIndex) -> None:
        """稀疏树上，严格约束满足不了，耗尽重试后抛 RuntimeError。"""
        from app.question_gen.sampler_v2 import sample_material_v2

        rng = random.Random(42)

        # VISUAL_FAMILY 要求 require_frames=True, min_l3_nodes=3
        # sparse_tree 只有 1 个 L3 且无帧 → 约束必然违反
        with pytest.raises(RuntimeError, match="max_attempts"):
            sample_material_v2(
                tree=sparse_tree,
                task_type="Object Recognition",
                used_node_ids=set(),
                rng=rng,
                level=3,
                constraint=VISUAL_FAMILY.sampling,
                max_attempts=3,
            )

    def test_cross_l2_populated_for_reasoning(self, real_tree: TreeIndex) -> None:
        """REASONING 家族要求 cross_l2_span=True，cross_l2_texts 应被填充。"""
        from app.question_gen.sampler_v2 import sample_material_v2

        rng = random.Random(42)

        result = sample_material_v2(
            tree=real_tree,
            task_type="Action Reasoning",
            used_node_ids=set(),
            rng=rng,
            level=2,
            constraint=REASONING_FAMILY.sampling,
        )

        # cross_l2_span=True 时必须有跨 L2 文本
        assert len(result.cross_l2_texts) > 0

    def test_subtitle_sentences_from_anchor(self, real_tree: TreeIndex) -> None:
        """采样结果的 subtitle_sentences 来自锚节点所属子树。"""
        from app.question_gen.sampler_v2 import sample_material_v2

        rng = random.Random(42)

        result = sample_material_v2(
            tree=real_tree,
            task_type="Action Reasoning",
            used_node_ids=set(),
            rng=rng,
            level=2,
            constraint=RETRIEVAL_FAMILY.sampling,
        )

        # real_tree 所有节点都有字幕，所以 subtitle_sentences 非空
        assert len(result.subtitle_sentences) > 0
        # 字幕应来自锚节点所属的子树（L2 自身字幕 + 子 L3 字幕）
        # fixture 中 L2 字幕格式: "L2事件字幕:L1={l1_idx}_L2={l2_idx}"
        # fixture 中 L3 字幕格式: "字幕L1={l1_idx}_L2={l2_idx}_L3={l3_idx}"
        # 解析锚 L2 的索引信息来验证
        anchor_l2_id = result.anchor.l2_id  # 如 "l1_1_l2_2"
        # 从 ID 提取 L1/L2 索引
        parts = anchor_l2_id.split("_")  # ["l1", "1", "l2", "2"]
        l1_idx, l2_idx = parts[1], parts[3]
        # 字幕中应包含 "L1={l1_idx}_L2={l2_idx}" 格式
        pattern = f"L1={l1_idx}_L2={l2_idx}"
        has_related = any(pattern in s for s in result.subtitle_sentences)
        assert has_related

    def test_invalid_level_raises(self, real_tree: TreeIndex) -> None:
        """无效的 level 参数应抛出 ValueError。"""
        from app.question_gen.sampler_v2 import sample_material_v2

        rng = random.Random(42)
        with pytest.raises(ValueError, match="level 必须为"):
            sample_material_v2(
                tree=real_tree,
                task_type="Object Recognition",
                used_node_ids=set(),
                rng=rng,
                level=4,
                constraint=RETRIEVAL_FAMILY.sampling,
            )


class TestValidateSamplingConstraints:
    """_validate_sampling_constraints 辅助函数测试。"""

    def test_passes_relaxed_constraint(self, real_tree: TreeIndex) -> None:
        """宽松约束在丰富树上应通过。"""
        from app.question_gen.sampler_v2 import _validate_sampling_constraints

        relaxed = SamplingConstraint(
            min_subtitles=1,
            min_l3_nodes=1,
            require_frames=False,
            cross_l2_span=False,
        )
        # 取第一个 L2 节点
        node_id = real_tree.roots[0].children[0].id
        assert _validate_sampling_constraints(real_tree, node_id, relaxed) is True

    def test_fails_strict_frame_constraint(self, sparse_tree: TreeIndex) -> None:
        """require_frames=True 但无帧时应返回 False。"""
        from app.question_gen.sampler_v2 import _validate_sampling_constraints

        strict = SamplingConstraint(
            min_subtitles=0,
            min_l3_nodes=1,
            require_frames=True,
            cross_l2_span=False,
        )
        node_id = sparse_tree.roots[0].children[0].id
        assert _validate_sampling_constraints(sparse_tree, node_id, strict) is False


class TestCollectSubtitleSentences:
    """_collect_subtitle_sentences 辅助函数测试。"""

    def test_collects_from_l2_and_l3(self, real_tree: TreeIndex) -> None:
        """收集指定节点的 L2 字幕和子 L3 字幕。"""
        from app.question_gen.sampler_v2 import _collect_subtitle_sentences

        l2_id = real_tree.roots[0].children[0].id
        sentences = _collect_subtitle_sentences(real_tree, (l2_id,))
        # 应包含 L2 自身字幕 + 3 个 L3 子节点字幕
        assert len(sentences) >= 1

    def test_empty_for_no_subtitles(self, sparse_tree: TreeIndex) -> None:
        """无字幕节点返回空列表。"""
        from app.question_gen.sampler_v2 import _collect_subtitle_sentences

        l2_id = sparse_tree.roots[0].children[0].id
        sentences = _collect_subtitle_sentences(sparse_tree, (l2_id,))
        assert sentences == []


class TestCollectCrossL2Context:
    """_collect_cross_l2_context 辅助函数测试。"""

    def test_returns_peer_l2_descriptions(self, real_tree: TreeIndex) -> None:
        """跨 L2 上下文应返回同 L1 下其他 L2 的描述。"""
        from app.question_gen.sampler_v2 import _collect_cross_l2_context

        anchor_l2_id = real_tree.roots[0].children[0].id
        texts = _collect_cross_l2_context(real_tree, anchor_l2_id, max_peers=3)
        # L1_0 有 3 个 L2，排除 anchor 后剩 2 个
        assert len(texts) == 2

    def test_max_peers_limits_output(self, real_tree: TreeIndex) -> None:
        """max_peers 参数限制返回数量。"""
        from app.question_gen.sampler_v2 import _collect_cross_l2_context

        anchor_l2_id = real_tree.roots[0].children[0].id
        texts = _collect_cross_l2_context(real_tree, anchor_l2_id, max_peers=1)
        assert len(texts) <= 1