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

"""轻量四门质量检查单元测试。"""

from __future__ import annotations

import json
from typing import Any

import pytest

from app.question_gen.gates import (
    CandidateQuestion,
    GateVerdict,
    _gate_blind_answer,
    _gate_key_verify,
    _gate_leak_test,
    _gate_multi_true,
    run_gates,
)
from app.question_gen.postprocess import PostprocessResult
from app.tree.index import (
    IndexMeta,
    L1Card,
    L1Node,
    L2Card,
    L2Node,
    L3Card,
    L3Node,
    TreeIndex,
)
from core.types import LLMResponse

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


def _make_llm_response(verdict: str, reason: str) -> LLMResponse:
    """构造一个 LLM 返回值，content 为标准 JSON 格式。"""
    return LLMResponse(
        content=json.dumps({"verdict": verdict, "reason": reason}),
        thinking="",
        model="mock-model",
        provider="mock",
        prompt_tokens=10,
        completion_tokens=5,
        latency_ms=100,
        ttft_ms=None,
        max_inter_token_ms=None,
        cache_hit=False,
        call_id="mock-call-001",
    )


class MockLLM:
    """可配置的 LLM mock — 按调用顺序返回预设响应。"""

    def __init__(self, responses: list[LLMResponse]) -> None:
        self._responses = list(responses)
        self._call_count = 0
        self.calls: list[dict[str, Any]] = []

    async def chat(
        self,
        messages: list[dict[str, Any]],
        *,
        session_id: str | None = None,
        parent_call_id: str | None = None,
    ) -> LLMResponse:
        """记录调用并返回预设响应。"""
        self.calls.append({"messages": messages, "session_id": session_id})
        idx = self._call_count
        self._call_count += 1
        if idx < len(self._responses):
            return self._responses[idx]
        return _make_llm_response("pass", "default")

    async def chat_with_images(
        self,
        messages: list[dict[str, Any]],
        images: list[Any],
        *,
        session_id: str | None = None,
        parent_call_id: str | None = None,
    ) -> LLMResponse:
        """记录 VLM 调用并返回预设响应（与 chat 共享计数器）。"""
        self.calls.append({"messages": messages, "images": images, "session_id": session_id})
        idx = self._call_count
        self._call_count += 1
        if idx < len(self._responses):
            return self._responses[idx]
        return _make_llm_response("pass", "default")


@pytest.fixture()
def mock_llm_pass() -> MockLLM:
    """返回始终 PASS 的 mock LLM。"""
    return MockLLM([_make_llm_response("pass", "evidence found")])


@pytest.fixture()
def mock_llm_fail() -> MockLLM:
    """返回始终 FAIL 的 mock LLM。"""
    return MockLLM([_make_llm_response("fail", "no evidence")])


def _make_candidate() -> CandidateQuestion:
    """构造测试用候选题目。"""
    return CandidateQuestion(
        question_id="q-001",
        video_id="v-001",
        task_type="Action Reasoning",
        skill_target="M1",
        question="What does the person do after picking up the book?",
        options=(
            "A. Reads it",
            "B. Puts it back",
            "C. Throws it away",
            "D. Gives it to someone",
        ),
        answer="A",
        source_nodes=("L2_001", "L3_001"),
        difficulty="medium",
    )


def _make_tree() -> TreeIndex:
    """构造最小测试树。"""
    l3_card = L3Card(
        frame_summary="Person picks up a book from the shelf and starts reading it.",
        visible_entities=["person", "book", "shelf"],
        ongoing_actions=["picking up book", "reading"],
        visible_text=[],
        spatial_layout="person in center, shelf on left",
        visual_attributes={},
        subtitle="He picks up the book and reads.",
    )
    l3 = L3Node(id="L3_001", card=l3_card)

    l2_card = L2Card(
        event_description="A person picks up a book from the shelf and begins reading it attentively.",
        entities=["person", "book"],
        actions=["pick up", "read"],
        action_subjects=["person"],
        visible_text=[],
        spatial_relations="person near shelf",
        state_changes="book moves from shelf to hands",
        subtitle="He picks up the book and reads.",
    )
    l2 = L2Node(id="L2_001", card=l2_card, children=[l3])

    l1_card = L1Card(
        scene_summary="Library scene with a person browsing and reading books.",
        main_setting="library",
        key_entities=["person", "books", "shelf"],
        main_actions=["browsing", "reading"],
        topic_keywords=["library", "reading"],
        visible_text=[],
        temporal_flow="enter → browse → pick up → read",
    )
    l1 = L1Node(id="L1_001", card=l1_card, children=[l2])

    meta = IndexMeta(source_path="test.mp4", modality="video")
    return TreeIndex(metadata=meta, roots=[l1])


def _make_postprocess(verbatim_ratio: float = 0.1) -> PostprocessResult:
    """构造测试用后处理结果。"""
    return PostprocessResult(
        options=(
            "A. Reads it",
            "B. Puts it back",
            "C. Throws it away",
            "D. Gives it to someone",
        ),
        answer="A",
        referent_violations=[],
        verbatim_ratio=verbatim_ratio,
        has_time_anchor=False,
    )


# ---------------------------------------------------------------------------
# TestGateKeyVerify
# ---------------------------------------------------------------------------


class TestGateKeyVerify:
    """key_verify 门：验证答案在来源素材中有证据支撑。"""

    @pytest.mark.asyncio()
    async def test_pass_evidence(self, mock_llm_pass: MockLLM) -> None:
        """LLM 确认有证据 → PASS。"""
        candidate = _make_candidate()
        tree = _make_tree()
        result = await _gate_key_verify(candidate, tree, mock_llm_pass, session_id="test-session")
        assert result.verdict == GateVerdict.PASS
        assert mock_llm_pass.calls  # 确认调用了 LLM

    @pytest.mark.asyncio()
    async def test_fail_no_evidence(self, mock_llm_fail: MockLLM) -> None:
        """LLM 判断无证据 → FAIL。"""
        candidate = _make_candidate()
        tree = _make_tree()
        result = await _gate_key_verify(candidate, tree, mock_llm_fail, session_id="test-session")
        assert result.verdict == GateVerdict.FAIL
        assert "no evidence" in result.reason


# ---------------------------------------------------------------------------
# TestGateBlindAnswer
# ---------------------------------------------------------------------------


class TestGateBlindAnswer:
    """blind_answer 门：无上下文答对 → FAIL（题目太简单/泄漏）。"""

    @pytest.mark.asyncio()
    async def test_pass_wrong(self) -> None:
        """LLM 在无上下文时答错 → PASS（题目确实需要视频信息）。"""
        # LLM 返回 "pass" 表示它无法正确回答
        llm = MockLLM([_make_llm_response("pass", "cannot determine without context")])
        candidate = _make_candidate()
        result = await _gate_blind_answer(candidate, llm, session_id="test-session")
        assert result.verdict == GateVerdict.PASS

    @pytest.mark.asyncio()
    async def test_fail_correct(self) -> None:
        """LLM 在无上下文时答对 → FAIL（题目泄漏）。"""
        llm = MockLLM([_make_llm_response("fail", "answer is obvious from options")])
        candidate = _make_candidate()
        result = await _gate_blind_answer(candidate, llm, session_id="test-session")
        assert result.verdict == GateVerdict.FAIL


# ---------------------------------------------------------------------------
# TestGateMultiTrue
# ---------------------------------------------------------------------------


class TestGateMultiTrue:
    """multi_true 门：多选项正确 → FAIL。"""

    @pytest.mark.asyncio()
    async def test_pass_single(self) -> None:
        """只有一个正确选项 → PASS。"""
        llm = MockLLM([_make_llm_response("pass", "only one correct answer")])
        candidate = _make_candidate()
        tree = _make_tree()
        result = await _gate_multi_true(candidate, tree, llm, session_id="test-session")
        assert result.verdict == GateVerdict.PASS

    @pytest.mark.asyncio()
    async def test_fail_multi(self) -> None:
        """多个选项可被视为正确 → FAIL。"""
        llm = MockLLM([_make_llm_response("fail", "options A and B are both plausible")])
        candidate = _make_candidate()
        tree = _make_tree()
        result = await _gate_multi_true(candidate, tree, llm, session_id="test-session")
        assert result.verdict == GateVerdict.FAIL


# ---------------------------------------------------------------------------
# TestGateLeakTest
# ---------------------------------------------------------------------------


class TestGateLeakTest:
    """leak_test 门：按家族模板执行泄漏探测。"""

    @pytest.mark.asyncio()
    async def test_per_strategy_template(self) -> None:
        """使用策略特定的 probe_template 调用 LLM。"""
        llm = MockLLM([_make_llm_response("pass", "no shortcut detected")])
        candidate = _make_candidate()
        result = await _gate_leak_test(
            candidate, "gate_leak_retrieval.md", llm, session_id="test-session"
        )
        assert result.verdict == GateVerdict.PASS
        # 验证 session_id 被正确传递
        assert llm.calls[0]["session_id"] == "test-session"


# ---------------------------------------------------------------------------
# TestRunGates
# ---------------------------------------------------------------------------


class TestRunGates:
    """run_gates 编排：并发四门 + verbatim 短路。"""

    @pytest.mark.asyncio()
    async def test_all_pass(self) -> None:
        """四门全 PASS → GateReport.passed=True。"""
        llm = MockLLM([_make_llm_response("pass", f"gate {i} ok") for i in range(4)])
        candidate = _make_candidate()
        tree = _make_tree()
        postprocess = _make_postprocess(verbatim_ratio=0.1)

        report = await run_gates(
            candidate=candidate,
            tree=tree,
            llm=llm,
            leak_probe_template="gate_leak_retrieval.md",
            postprocess=postprocess,
            session_id="test-session",
        )
        assert report.passed is True
        assert report.reject_reason is None

    @pytest.mark.asyncio()
    async def test_high_verbatim_shortcircuits(self) -> None:
        """verbatim_ratio > 0.5 → key_verify 直接 FAIL，其余 SKIP，不调用 LLM。"""
        llm = MockLLM([_make_llm_response("pass", "should not be called")])
        candidate = _make_candidate()
        tree = _make_tree()
        postprocess = _make_postprocess(verbatim_ratio=0.8)

        report = await run_gates(
            candidate=candidate,
            tree=tree,
            llm=llm,
            leak_probe_template="gate_leak_retrieval.md",
            postprocess=postprocess,
            session_id="test-session",
        )
        assert report.passed is False
        assert report.key_verify.verdict == GateVerdict.FAIL
        assert "verbatim" in report.key_verify.reason.lower()
        # 其余三门应为 SKIP
        assert report.blind_answer.verdict == GateVerdict.SKIP
        assert report.multi_true.verdict == GateVerdict.SKIP
        assert report.leak_test.verdict == GateVerdict.SKIP
        # LLM 不应被调用
        assert len(llm.calls) == 0