"""QuestionGenStore 单元测试 — SQLite 出题管线日志记录器。 覆盖 schema 幂等性、run 生命周期、item 写入与门判定更新、难度更新。 """ from __future__ import annotations from dataclasses import dataclass from enum import Enum from typing import TYPE_CHECKING import pytest if TYPE_CHECKING: from pathlib import Path from app.question_gen.run_store import QuestionGenStore, RunStats # --------------------------------------------------------------------------- # GateReport 测试替身（Task 6 尚未实现，此处构造兼容 duck-type） # --------------------------------------------------------------------------- class _Verdict(Enum): """门判定枚举替身。""" PASS = "pass" FAIL = "fail" SKIP = "skip" @dataclass(frozen=True) class _GateResult: """单门判定结果替身。""" verdict: _Verdict reason: str @dataclass class _MockGateReport: """GateReport 测试替身，duck-type 兼容。""" key_verify: _GateResult blind_answer: _GateResult multi_true: _GateResult leak_test: _GateResult @property def passed(self) -> bool: """所有门均 PASS 或 SKIP 时视为通过。""" results = [self.key_verify, self.blind_answer, self.multi_true, self.leak_test] return all(r.verdict != _Verdict.FAIL for r in results) @property def reject_reason(self) -> str | None: """返回首个 FAIL 门的 reason，若全部通过则 None。""" for r in [self.key_verify, self.blind_answer, self.multi_true, self.leak_test]: if r.verdict == _Verdict.FAIL: return r.reason return None # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def store(tmp_path: Path) -> QuestionGenStore: """创建临时 SQLite 上的 QuestionGenStore 实例。""" db_path = tmp_path / "question_gen.db" s = QuestionGenStore(db_path=db_path) yield s s.close() # --------------------------------------------------------------------------- # TestQuestionGenStore # --------------------------------------------------------------------------- class TestQuestionGenStore: """QuestionGenStore 全功能验证。""" def test_schema_idempotent(self, tmp_path: Path) -> None: """连续两次初始化同一 db 不报错，表结构不变。""" db_path = tmp_path / "idem.db" s1 = QuestionGenStore(db_path=db_path) s2 = QuestionGenStore(db_path=db_path) # 第二次初始化不应抛异常 s2.record_run_start(run_id="r-idem", git_sha="abc123", config_snapshot="{}") s1.close() s2.close() def test_record_run_lifecycle(self, store: QuestionGenStore) -> None: """start -> record items -> end -> get_stats 完整生命周期。""" run_id = "run-001" store.record_run_start( run_id=run_id, git_sha="deadbeef", config_snapshot='{"n_slots": 240}', ) # 写入几个 item 来模拟统计 for i in range(5): store.record_item( item_id=f"item-{i}", run_id=run_id, slot_id=f"slot-{i}", video_id="v001", family="retrieval", task_type="Object Recognition", skill_target="M1", attempt=1, question_text=f"问题{i}", ) # 结束 run stats = RunStats(total_slots=10, accepted=5, rejected=3, heavy_sampled=2) store.record_run_end(run_id=run_id, status="completed", stats=stats) # 查询 stats retrieved = store.get_run_stats(run_id) assert retrieved.total_slots == 10 assert retrieved.accepted == 5 assert retrieved.rejected == 3 assert retrieved.heavy_sampled == 2 def test_record_item_and_gates(self, store: QuestionGenStore) -> None: """record_item 后 update_gates 正确更新门判定列与 final_status。""" run_id = "run-gates" store.record_run_start(run_id=run_id, git_sha="aaa111", config_snapshot="{}") item_id = "item-gate-1" store.record_item( item_id=item_id, run_id=run_id, slot_id="slot-0", video_id="v002", family="reasoning", task_type="Action Sequence", skill_target="M2", attempt=1, question_text="为什么这样做？", ) # 所有门通过 report_pass = _MockGateReport( key_verify=_GateResult(_Verdict.PASS, "ok"), blind_answer=_GateResult(_Verdict.PASS, "ok"), multi_true=_GateResult(_Verdict.SKIP, "n/a"), leak_test=_GateResult(_Verdict.PASS, "ok"), ) store.update_gates(item_id=item_id, report=report_pass) # 验证数据库中的值 import sqlite3 conn = sqlite3.connect(str(store._db_path)) conn.row_factory = sqlite3.Row row = conn.execute( "SELECT * FROM question_gen_items WHERE item_id=?", (item_id,) ).fetchone() conn.close() assert row["gate_key_verify"] == "pass" assert row["gate_blind_answer"] == "pass" assert row["gate_multi_true"] == "skip" assert row["gate_leak_test"] == "pass" assert row["gate_reject_reason"] is None assert row["final_status"] == "accepted" def test_update_gates_rejected(self, store: QuestionGenStore) -> None: """门判定失败时 final_status 为 rejected 且记录 reject_reason。""" run_id = "run-rej" store.record_run_start(run_id=run_id, git_sha="bbb222", config_snapshot="{}") item_id = "item-rej-1" store.record_item( item_id=item_id, run_id=run_id, slot_id="slot-1", video_id="v003", family="visual", task_type="Scene Transition", skill_target="M3", attempt=2, question_text="画面中有什么？", ) report_fail = _MockGateReport( key_verify=_GateResult(_Verdict.PASS, "ok"), blind_answer=_GateResult(_Verdict.FAIL, "答案可由常识推断"), multi_true=_GateResult(_Verdict.SKIP, "n/a"), leak_test=_GateResult(_Verdict.SKIP, "n/a"), ) store.update_gates(item_id=item_id, report=report_fail) import sqlite3 conn = sqlite3.connect(str(store._db_path)) conn.row_factory = sqlite3.Row row = conn.execute( "SELECT * FROM question_gen_items WHERE item_id=?", (item_id,) ).fetchone() conn.close() assert row["final_status"] == "rejected" assert row["gate_reject_reason"] == "答案可由常识推断" def test_update_difficulty(self, store: QuestionGenStore) -> None: """update_difficulty 正确写入 difficulty_steps。""" run_id = "run-diff" store.record_run_start(run_id=run_id, git_sha="ccc333", config_snapshot="{}") item_id = "item-diff-1" store.record_item( item_id=item_id, run_id=run_id, slot_id="slot-2", video_id="v004", family="enumeration", task_type="Counting", skill_target="M4", attempt=1, question_text="有多少个？", ) store.update_difficulty(item_id=item_id, difficulty_steps=7) import sqlite3 conn = sqlite3.connect(str(store._db_path)) conn.row_factory = sqlite3.Row row = conn.execute( "SELECT difficulty_steps FROM question_gen_items WHERE item_id=?", (item_id,), ).fetchone() conn.close() assert row["difficulty_steps"] == 7 def test_get_run_stats_not_found(self, store: QuestionGenStore) -> None: """查询不存在的 run_id 应报错。""" with pytest.raises(ValueError, match="run_id"): store.get_run_stats("nonexistent-run")