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

"""selector 织入辅助：正解文本提取 + 分流。"""

from pathlib import Path

import pytest

from app.question_gen.pipeline_v2 import _extract_correct_text


def test_extract_correct_text_strips_prefix():
    options = ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
    assert _extract_correct_text(options, "C") == "煮"


def test_extract_correct_text_handles_lowercase_answer():
    options = ("A. run", "B. walk", "C. jump", "D. sit")
    assert _extract_correct_text(options, "b") == "walk"


def test_extract_correct_text_out_of_range_raises():
    options = ("A. a", "B. b", "C. c", "D. d")
    with pytest.raises(ValueError):
        _extract_correct_text(options, "E")


def test_replace_candidate_options():
    from app.question_gen.generator_v2 import CandidateQuestion
    from app.question_gen.pipeline_v2 import _replace_candidate_options

    c = CandidateQuestion(
        question_id="q",
        video_id="v",
        task_type="Action Recognition",
        skill_target="M1_AR",
        question="?",
        options=("A. a", "B. b", "C. c", "D. d"),
        answer="A",
        source_nodes=("n1",),
        difficulty="hard",
    )
    new = _replace_candidate_options(c, ("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), "A")
    assert new.options == ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
    assert new.question == "?"  # 其余字段不变
    assert new.source_nodes == ("n1",)


class _RecordingStore:
    """记录 store 调用的假实现，用于断言落库一致性。"""

    def __init__(self) -> None:
        self.rejected: list[tuple[str, str]] = []
        self.selector_scores: list[tuple[str, str]] = []

    def mark_item_rejected(self, item_id: str, reason: str) -> None:
        self.rejected.append((item_id, reason))

    def update_selector_scores(self, item_id: str, selector_scores_json: str) -> None:
        self.selector_scores.append((item_id, selector_scores_json))


class _ARStrategy:
    """最小策略替身：仅暴露 uses_grounded_selector=True。"""

    uses_grounded_selector = True


class _DummyMaterial:
    """占位素材：异常在 build_grounded_options 内抛出前从不被读取。"""


async def test_apply_grounded_selector_marks_rejected_on_error(monkeypatch):
    """selector 异常路径须 mark_item_rejected（与 hard-fail 落库一致），但不写 selector_scores。"""
    import app.question_gen.distractor_selector as ds
    from app.question_gen.generator_v2 import CandidateQuestion
    from app.question_gen.pipeline_v2 import PipelineConfig, _apply_grounded_selector

    async def _raise(**_kwargs):
        raise ValueError("boom")

    monkeypatch.setattr(ds, "build_grounded_options", _raise)

    candidate = CandidateQuestion(
        question_id="q",
        video_id="v",
        task_type="Action Recognition",
        skill_target="M1_AR",
        question="?",
        options=("A. a", "B. b", "C. c", "D. d"),
        answer="A",
        source_nodes=("n1",),
        difficulty="hard",
    )
    config = PipelineConfig(
        per_type=1,
        retry_limit=1,
        heavy_sample_rate=0.0,
        dedup_threshold=0.85,
        concurrency=1,
        seed=0,
        output_dir=Path("."),
    )
    store = _RecordingStore()

    result_candidate, reason = await _apply_grounded_selector(
        candidate,
        _ARStrategy(),
        _DummyMaterial(),
        vlm=None,
        config=config,
        store=store,
        item_id="item-1",
        slot_id="slot-1",
        attempt=1,
        session_id="s",
    )

    assert result_candidate is None
    assert reason is not None
    assert reason.startswith("selector_error: ")
    assert store.rejected == [("item-1", reason)]
    # 异常路径无 observation，故不落 selector_scores
    assert store.selector_scores == []


@pytest.mark.parametrize(
    "exc",
    [
        RuntimeError("vlm 500"),
        TimeoutError("watchdog total timeout"),
        ConnectionError("network reset"),
    ],
)
async def test_apply_grounded_selector_isolates_vlm_runtime_exceptions(monkeypatch, exc):
    """C3: selector 内 VLM 网络/超时/熔断（非 ValueError）异常须被隔离为 selector_error 重出，

    不得穿透 _process_one_slot → asyncio.gather 崩掉整批生成。
    """
    import app.question_gen.distractor_selector as ds
    from app.question_gen.generator_v2 import CandidateQuestion
    from app.question_gen.pipeline_v2 import PipelineConfig, _apply_grounded_selector

    async def _raise(**_kwargs):
        raise exc

    monkeypatch.setattr(ds, "build_grounded_options", _raise)

    candidate = CandidateQuestion(
        question_id="q",
        video_id="v",
        task_type="Action Recognition",
        skill_target="M1_AR",
        question="?",
        options=("A. a", "B. b", "C. c", "D. d"),
        answer="A",
        source_nodes=("n1",),
        difficulty="hard",
    )
    config = PipelineConfig(
        per_type=1,
        retry_limit=1,
        heavy_sample_rate=0.0,
        dedup_threshold=0.85,
        concurrency=1,
        seed=0,
        output_dir=Path("."),
    )
    store = _RecordingStore()

    result_candidate, reason = await _apply_grounded_selector(
        candidate,
        _ARStrategy(),
        _DummyMaterial(),
        vlm=None,
        config=config,
        store=store,
        item_id="item-1",
        slot_id="slot-1",
        attempt=1,
        session_id="s",
    )

    assert result_candidate is None
    assert reason is not None
    assert reason.startswith("selector_error: ")
    assert store.rejected == [("item-1", reason)]
    assert store.selector_scores == []