164 lines
6.1 KiB
Python
164 lines
6.1 KiB
Python
"""build_split 端到端集成测试：真实 infer_adhoc 预测 + 构造诊断信号 → 冻结三池。
|
||
|
||
用 workspaces/default/harness.db 的真实 infer_adhoc 基线预测（900 题 / 300 视频，
|
||
每视频 3 题）驱动结果驱动视频级切分，构造一份 T2 诊断信号 store 喂给选择器，
|
||
验证冻结的 pools.json 满足全部防御断言（三池视频互斥、覆盖 900 题、每视频 3 题、
|
||
内容指纹一致），确保 capstone 编排把前面所有组件正确串联。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import sqlite3
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore
|
||
from app.harness.build_split import SplitBuildConfig, build_split
|
||
from app.harness.split_selection import evolution_target_of
|
||
from core.evolution.types import DiagnosisSignalRow
|
||
|
||
_HARNESS_DB = Path("workspaces/default/harness.db")
|
||
_BENCHMARK_DIR = Path("store/questions/benchmarks/Video-MME")
|
||
_BASELINE_RUN_ID = "infer_adhoc"
|
||
_DIAG_FINGERPRINT = "diag_test_v1"
|
||
# 四类错误类别轮转，确定性铺满多样性格子（cell = task_type × error_type）。
|
||
_ERROR_TYPES = ("extraction_failure", "search_failure", "reasoning_failure", "mixed")
|
||
|
||
|
||
def _normalize(choice: str | None) -> str:
|
||
"""选项归一：strip → 大写 → 取首字母（None 归一为空串）。"""
|
||
return (choice or "").strip().upper()[:1]
|
||
|
||
|
||
def _read_canonical_predictions() -> list[dict]:
|
||
"""从真实 harness.db 读 infer_adhoc 每题首行（ORDER BY rowid）作为 canonical 预测。"""
|
||
conn = sqlite3.connect(f"file:{_HARNESS_DB}?mode=ro", uri=True)
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute(
|
||
"SELECT question_id, video_id, task_type, prediction, answer "
|
||
"FROM predictions WHERE run_id = ? ORDER BY rowid",
|
||
(_BASELINE_RUN_ID,),
|
||
).fetchall()
|
||
conn.close()
|
||
seen: dict[str, dict] = {}
|
||
for row in rows:
|
||
qid = row["question_id"]
|
||
if qid in seen:
|
||
continue
|
||
seen[qid] = {
|
||
"question_id": qid,
|
||
"video_id": row["video_id"],
|
||
"task_type": row["task_type"],
|
||
"correct": _normalize(row["prediction"]) == _normalize(row["answer"]),
|
||
}
|
||
return list(seen.values())
|
||
|
||
|
||
def _populate_signal_store(store: SqliteDiagnosisSignalStore, preds: list[dict]) -> None:
|
||
"""为全部错题写入 T2 诊断信号（error_type 轮转），构造可训练缺陷多样性。"""
|
||
wrong = [p for p in preds if not p["correct"]]
|
||
for idx, pred in enumerate(wrong):
|
||
error_type = _ERROR_TYPES[idx % len(_ERROR_TYPES)]
|
||
store.upsert(
|
||
DiagnosisSignalRow(
|
||
question_id=pred["question_id"],
|
||
video_id=pred["video_id"],
|
||
baseline_run_id=_BASELINE_RUN_ID,
|
||
diag_fingerprint=_DIAG_FINGERPRINT,
|
||
task_type=pred["task_type"],
|
||
error_type=error_type,
|
||
cause_category="defect",
|
||
tier="T2",
|
||
evolution_target=evolution_target_of(error_type),
|
||
degraded=False,
|
||
infra=False,
|
||
session_id=None,
|
||
)
|
||
)
|
||
|
||
|
||
@pytest.mark.skipif(
|
||
not _HARNESS_DB.exists() or not _BENCHMARK_DIR.exists(),
|
||
reason="需要真实 workspaces/default/harness.db 与 Video-MME benchmark",
|
||
)
|
||
def test_end_to_end_freezes_valid_pools(tmp_path: Path) -> None:
|
||
"""真实基线预测 + 构造 T2 信号 → 冻结 pools.json，校验六条防御断言。"""
|
||
out = tmp_path / "pools.json"
|
||
manifest_path = tmp_path / "split_manifest.json"
|
||
signal_db = tmp_path / "signals.db"
|
||
|
||
preds = _read_canonical_predictions()
|
||
store = SqliteDiagnosisSignalStore(str(signal_db))
|
||
_populate_signal_store(store, preds)
|
||
|
||
config = SplitBuildConfig(
|
||
n_trainval=100,
|
||
floor_k={"Object Reasoning": 3},
|
||
epsilon=0.1,
|
||
report_floor=30,
|
||
select_seed=7,
|
||
val_ratio=0.3,
|
||
split_seed=7,
|
||
val_wrong_min=0,
|
||
)
|
||
|
||
result = build_split(
|
||
db_path=_HARNESS_DB,
|
||
baseline_run_id=_BASELINE_RUN_ID,
|
||
signal_store=store,
|
||
diag_fingerprint=_DIAG_FINGERPRINT,
|
||
questions_dir=_BENCHMARK_DIR,
|
||
config=config,
|
||
out_path=out,
|
||
manifest_path=manifest_path,
|
||
generated_at="2026-07-15T00:00:00Z",
|
||
)
|
||
store.close()
|
||
|
||
pools = result["pools"]
|
||
|
||
# 防御① 三池视频集两两不相交
|
||
diag_v = {q.video_id for q in pools.diagnosis}
|
||
val_v = {q.video_id for q in pools.validation}
|
||
test_v = {q.video_id for q in pools.test}
|
||
assert diag_v & test_v == set()
|
||
assert val_v & test_v == set()
|
||
assert diag_v & val_v == set()
|
||
|
||
# 防御② 三池覆盖 900 题（按 distinct question）
|
||
all_q = pools.diagnosis + pools.validation + pools.test
|
||
assert len({q.question_id for q in all_q}) == 900
|
||
assert len(all_q) == 900
|
||
|
||
# 防御③ 每 video 恰 3 题
|
||
per_video: dict[str, int] = {}
|
||
for q in all_q:
|
||
per_video[q.video_id] = per_video.get(q.video_id, 0) + 1
|
||
assert set(per_video.values()) == {3}
|
||
|
||
# 防御⑤ manifest 的 pools_sha256 == sha256(out 文件内容)
|
||
assert (
|
||
result["manifest"]["pools_sha256"]
|
||
== hashlib.sha256(out.read_text(encoding="utf-8").encode("utf-8")).hexdigest()
|
||
)
|
||
|
||
# assignment 覆盖全部 300 视频且取值合法
|
||
assignment = result["assignment"]
|
||
assert set(assignment.values()) <= {"trainval", "test"}
|
||
assert len(assignment) == 300
|
||
|
||
# manifest 覆盖报告含关键指标
|
||
coverage = result["manifest"]["coverage_report"]
|
||
assert coverage["grid_total"] == 48
|
||
assert "tier_distribution" in coverage
|
||
assert manifest_path.exists()
|
||
|
||
# I-1：evolution_target_distribution 出现在 coverage_report，且 T2 信号按 tool/skill/system 计数。
|
||
target_dist = coverage["evolution_target_distribution"]
|
||
assert set(target_dist) <= {"tool", "skill", "system"}
|
||
# 全部错题构造为 T2（error_type 四类轮转），进化目标覆盖三层且计数为正。
|
||
assert sum(target_dist.values()) > 0
|
||
assert set(target_dist) == {"tool", "skill", "system"}
|