iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add v3 question-gen observation tables to run_store

Browse Source
This commit is contained in:
iomgaa
2026-07-15 09:33:40 -04:00
parent 1ecb6ba25c
commit 5ecbac620c
2 changed files with 770 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_run_store_v3_tables.py
+367
View File
@@ -0,0 +1,367 @@
"""v3 出题领域观测表落库测试 — facts/unit_verdict/collapse_metrics/quarantine/resume_state。
验证 5 张 v3 表的 insert 方法落库+读回、幂等 upsert（同指纹/同 unit_id 不产生重复行）、
以及 ts 由外部传入（禁进程内 now，保幂等可复现）。同时确认 v2 旧表未受本 Task 影响。
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from app.question_gen.run_store import QuestionGenStore, RunStats
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture
def store(tmp_path: Path) -> QuestionGenStore:
"""基于真临时 SQLite 文件构造 QuestionGenStore。"""
db_path = tmp_path / "qgen_v3.db"
s = QuestionGenStore(db_path)
yield s
s.close()
# ---------------------------------------------------------------------------
# 表 1：facts
# ---------------------------------------------------------------------------
def test_insert_fact_roundtrip(store: QuestionGenStore) -> None:
"""insert_fact 落库后可读回全部列，ts 为外部传入值。"""
ts = "2026-07-15T10:00:00+00:00"
store.insert_fact(
fact_id="fact-1",
video_id="vid-1",
segment_id="seg-1",
subject="man",
action="pour",
object="water",
frame_ids="[10, 20, 30]",
polarity="真",
fact_type="binding",
difficulty_tier=2,
verifier_refs='{"qwen": "pass", "minimax": "pass"}',
cross_agree=1,
negative_at_target="not_false",
session_id="sess-1",
ts=ts,
)
row = store._conn.execute(
"SELECT fact_id, video_id, segment_id, subject, action, object, frame_ids, "
"polarity, fact_type, difficulty_tier, verifier_refs, cross_agree, "
"negative_at_target, session_id, ts FROM facts WHERE fact_id='fact-1'"
).fetchone()
assert row == (
"fact-1",
"vid-1",
"seg-1",
"man",
"pour",
"water",
"[10, 20, 30]",
"真",
"binding",
2,
'{"qwen": "pass", "minimax": "pass"}',
1,
"not_false",
"sess-1",
ts,
)
def test_insert_fact_ts_is_external_not_now(store: QuestionGenStore) -> None:
"""ts 如实存储外部传入的历史时间戳，不被进程内 now 覆盖。"""
old_ts = "2020-01-01T00:00:00+00:00"
store.insert_fact(
fact_id="fact-old",
video_id="v",
segment_id="s",
subject="a",
action="b",
object="c",
frame_ids="[]",
polarity="真",
fact_type="state",
difficulty_tier=1,
verifier_refs="{}",
cross_agree=0,
negative_at_target="x",
session_id="sess",
ts=old_ts,
)
stored = store._conn.execute("SELECT ts FROM facts WHERE fact_id='fact-old'").fetchone()[0]
assert stored == old_ts
# ---------------------------------------------------------------------------
# 表 2：unit_verdict
# ---------------------------------------------------------------------------
def test_insert_unit_verdict_roundtrip(store: QuestionGenStore) -> None:
"""insert_unit_verdict 落库后可读回，含可空 pair_id/metric_value/model。"""
ts = "2026-07-15T11:00:00+00:00"
store.insert_unit_verdict(
unit_id="unit-1",
pair_id="pair-1",
sub_pattern="binding_swap",
stage=3,
verdict="pass",
reason="ok",
metric_value=0.87,
model="qwen-vl",
session_id="sess-1",
ts=ts,
)
row = store._conn.execute(
"SELECT unit_id, pair_id, sub_pattern, stage, verdict, reason, metric_value, "
"model, session_id, ts FROM unit_verdict WHERE unit_id='unit-1' AND stage=3"
).fetchone()
assert row == (
"unit-1",
"pair-1",
"binding_swap",
3,
"pass",
"ok",
0.87,
"qwen-vl",
"sess-1",
ts,
)
def test_insert_unit_verdict_nullable_fields(store: QuestionGenStore) -> None:
"""可空列（pair_id/metric_value/model）支持 None 落库。"""
store.insert_unit_verdict(
unit_id="unit-2",
pair_id=None,
sub_pattern="order",
stage=1,
verdict="abstain",
reason="no judge",
metric_value=None,
model=None,
session_id="sess-1",
ts="2026-07-15T11:05:00+00:00",
)
row = store._conn.execute(
"SELECT pair_id, metric_value, model FROM unit_verdict WHERE unit_id='unit-2'"
).fetchone()
assert row == (None, None, None)
def test_insert_unit_verdict_rerun_same_stage_overwrites(store: QuestionGenStore) -> None:
"""同 (unit_id, stage) 重跑覆盖更新，行数保持 1。"""
common = {
"unit_id": "unit-3",
"pair_id": "pair-3",
"sub_pattern": "manner",
"stage": 2,
"session_id": "sess-1",
}
store.insert_unit_verdict(
verdict="fail",
reason="first",
metric_value=0.1,
model="m1",
ts="2026-07-15T11:10:00+00:00",
**common,
)
store.insert_unit_verdict(
verdict="pass",
reason="second",
metric_value=0.9,
model="m2",
ts="2026-07-15T11:20:00+00:00",
**common,
)
rows = store._conn.execute(
"SELECT verdict, reason, metric_value FROM unit_verdict WHERE unit_id='unit-3' AND stage=2"
).fetchall()
assert len(rows) == 1
assert rows[0] == ("pass", "second", 0.9)
# ---------------------------------------------------------------------------
# 表 3：collapse_metrics
# ---------------------------------------------------------------------------
def test_insert_collapse_metrics_roundtrip(store: QuestionGenStore) -> None:
"""insert_collapse_metrics 落库后可读回全部度量列。"""
ts = "2026-07-15T12:00:00+00:00"
store.insert_collapse_metrics(
pair_id="pair-1",
text_only_acc=0.3,
single_frame_acc=0.4,
placebo_drop=0.1,
majority_vote_hit=0.25,
slot_chi2=1.5,
distractor_min_dist=0.6,
multiformat_consistency=0.8,
subtitle_answerability=0.2,
ts=ts,
)
row = store._conn.execute(
"SELECT pair_id, text_only_acc, single_frame_acc, placebo_drop, majority_vote_hit, "
"slot_chi2, distractor_min_dist, multiformat_consistency, subtitle_answerability, ts "
"FROM collapse_metrics WHERE pair_id='pair-1'"
).fetchone()
assert row == (
"pair-1",
0.3,
0.4,
0.1,
0.25,
1.5,
0.6,
0.8,
0.2,
ts,
)
def test_insert_collapse_metrics_recompute_upserts(store: QuestionGenStore) -> None:
"""同 pair_id 重算覆盖更新，行数保持 1。"""
store.insert_collapse_metrics(
pair_id="pair-2",
text_only_acc=0.3,
single_frame_acc=0.4,
placebo_drop=0.1,
majority_vote_hit=0.25,
slot_chi2=1.0,
distractor_min_dist=0.5,
multiformat_consistency=0.7,
subtitle_answerability=0.1,
ts="2026-07-15T12:00:00+00:00",
)
store.insert_collapse_metrics(
pair_id="pair-2",
text_only_acc=0.05,
single_frame_acc=0.05,
placebo_drop=0.5,
majority_vote_hit=0.25,
slot_chi2=9.0,
distractor_min_dist=0.9,
multiformat_consistency=0.95,
subtitle_answerability=0.0,
ts="2026-07-15T12:30:00+00:00",
)
rows = store._conn.execute(
"SELECT text_only_acc, slot_chi2 FROM collapse_metrics WHERE pair_id='pair-2'"
).fetchall()
assert len(rows) == 1
assert rows[0] == (0.05, 9.0)
# ---------------------------------------------------------------------------
# 表 4：quarantine
# ---------------------------------------------------------------------------
def test_quarantine_roundtrip(store: QuestionGenStore) -> None:
"""quarantine 落库后可读回，ts 为外部传入值。"""
ts = "2026-07-15T13:00:00+00:00"
store.quarantine(
content_fingerprint="fp-1",
sub_pattern="binding_swap",
quarantine_reason="double_true",
round_no=1,
ts=ts,
)
row = store._conn.execute(
"SELECT content_fingerprint, sub_pattern, quarantine_reason, round_no, ts "
"FROM quarantine WHERE content_fingerprint='fp-1'"
).fetchone()
assert row == ("fp-1", "binding_swap", "double_true", 1, ts)
def test_quarantine_same_fingerprint_dedup(store: QuestionGenStore) -> None:
"""同 content_fingerprint 重复调用不产生重复行（去重幂等），且内容被更新。"""
store.quarantine(
content_fingerprint="fp-dup",
sub_pattern="order",
quarantine_reason="reason_a",
round_no=1,
ts="2026-07-15T13:00:00+00:00",
)
store.quarantine(
content_fingerprint="fp-dup",
sub_pattern="order",
quarantine_reason="reason_b",
round_no=2,
ts="2026-07-15T13:30:00+00:00",
)
rows = store._conn.execute(
"SELECT quarantine_reason, round_no FROM quarantine WHERE content_fingerprint='fp-dup'"
).fetchall()
assert len(rows) == 1
assert rows[0] == ("reason_b", 2)
# ---------------------------------------------------------------------------
# 表 5：resume_state
# ---------------------------------------------------------------------------
def test_upsert_resume_state_roundtrip(store: QuestionGenStore) -> None:
"""upsert_resume_state 落库后可读回全部列。"""
store.upsert_resume_state(
unit_id="unit-1",
status="pending",
config_fingerprint="cfg-abc",
seq_offset=0,
)
row = store._conn.execute(
"SELECT unit_id, status, config_fingerprint, seq_offset "
"FROM resume_state WHERE unit_id='unit-1'"
).fetchone()
assert row == ("unit-1", "pending", "cfg-abc", 0)
def test_upsert_resume_state_same_unit_overwrites(store: QuestionGenStore) -> None:
"""同 unit_id 覆盖更新 status/config/seq_offset，行数保持 1。"""
store.upsert_resume_state(
unit_id="unit-2",
status="pending",
config_fingerprint="cfg-1",
seq_offset=0,
)
store.upsert_resume_state(
unit_id="unit-2",
status="accepted",
config_fingerprint="cfg-2",
seq_offset=5,
)
rows = store._conn.execute(
"SELECT status, config_fingerprint, seq_offset FROM resume_state WHERE unit_id='unit-2'"
).fetchall()
assert len(rows) == 1
assert rows[0] == ("accepted", "cfg-2", 5)
# ---------------------------------------------------------------------------
# v2 旧表未受影响
# ---------------------------------------------------------------------------
def test_v2_tables_unaffected(store: QuestionGenStore) -> None:
"""v3 建表后 v2 旧表仍存在且旧方法仍可用。"""
store.record_run_start(run_id="run-1", git_sha="deadbeef", config_snapshot="{}")
store.record_run_end(
run_id="run-1",
status="completed",
stats=RunStats(total_slots=10, accepted=8, rejected=2, heavy_sampled=1),
)
stats = store.get_run_stats("run-1")
assert stats.accepted == 8
tables = {
r[0] for r in store._conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
assert {"question_gen_runs", "question_gen_items", "adversarial_verdicts"} <= tables