99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
"""SqliteDiagnosisSignalStore 单元测试。
|
|
|
|
验证逐题诊断信号的 upsert 幂等、断点续跑（done_question_ids）、
|
|
load 往返还原，以及 INFRA 行的可空字段处理。
|
|
"""
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore
|
|
from core.evolution.types import DiagnosisSignalRow
|
|
|
|
|
|
def _store(tmp_path):
|
|
"""构造指向临时 harness.db 的信号存储。"""
|
|
return SqliteDiagnosisSignalStore(str(tmp_path / "h.db"))
|
|
|
|
|
|
def test_upsert_and_get_done_ids(tmp_path):
|
|
s = _store(tmp_path)
|
|
row = DiagnosisSignalRow(
|
|
question_id="q1",
|
|
video_id="v1",
|
|
baseline_run_id="r",
|
|
diag_fingerprint="fp",
|
|
task_type="Counting",
|
|
error_type="search_failure",
|
|
cause_category="defect",
|
|
tier="T2",
|
|
evolution_target="skill",
|
|
degraded=False,
|
|
infra=False,
|
|
session_id="s",
|
|
)
|
|
s.upsert(row)
|
|
assert s.done_question_ids("r", "fp") == {"q1"}
|
|
s.upsert(row) # 同键覆盖，不重复
|
|
assert s.done_question_ids("r", "fp") == {"q1"}
|
|
assert s.done_question_ids("r", "other_fp") == set() # 不同 fingerprint 隔离
|
|
|
|
|
|
def test_load_rows_roundtrip(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.upsert(
|
|
DiagnosisSignalRow(
|
|
"q2",
|
|
"v2",
|
|
"r",
|
|
"fp",
|
|
"OCR Problems",
|
|
"extraction_failure",
|
|
"lapse",
|
|
"T1",
|
|
"tool",
|
|
False,
|
|
False,
|
|
"s",
|
|
)
|
|
)
|
|
rows = s.load("r", "fp")
|
|
assert len(rows) == 1 and rows[0].tier == "T1" and rows[0].evolution_target == "tool"
|
|
|
|
|
|
def test_null_fields_for_infra_row(tmp_path):
|
|
s = _store(tmp_path)
|
|
# T0 INFRA 行：error_type/cause_category/evolution_target 为 None
|
|
s.upsert(
|
|
DiagnosisSignalRow(
|
|
"q3",
|
|
"v3",
|
|
"r",
|
|
"fp",
|
|
"Counting",
|
|
None,
|
|
None,
|
|
"T0",
|
|
None,
|
|
degraded=False,
|
|
infra=True,
|
|
session_id=None,
|
|
)
|
|
)
|
|
rows = s.load("r", "fp")
|
|
assert rows[0].error_type is None and rows[0].evolution_target is None and rows[0].infra is True
|
|
|
|
|
|
def test_context_manager_closes_connection(tmp_path):
|
|
# with 块退出后连接关闭，再操作应报 ProgrammingError
|
|
with SqliteDiagnosisSignalStore(str(tmp_path / "h.db")) as s:
|
|
s.upsert(
|
|
DiagnosisSignalRow(
|
|
"q4", "v4", "r", "fp", "Counting", None, None, "T0", None, False, True, None
|
|
)
|
|
)
|
|
assert s.done_question_ids("r", "fp") == {"q4"}
|
|
with pytest.raises(sqlite3.ProgrammingError):
|
|
s.done_question_ids("r", "fp")
|