946 lines
32 KiB
Python
946 lines
32 KiB
Python
"""三池切分单元测试。
|
||
|
||
验证:
|
||
- 三池互斥（question_id 无重叠）
|
||
- test 池自然分布（correct_ratio=None）
|
||
- save/load 往返一致
|
||
- 旧格式拒绝（无 test 键 → ValueError）
|
||
- build_or_load_pools 冻结复用（pools.json 存在时不重切）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import TYPE_CHECKING
|
||
|
||
import pytest
|
||
|
||
from app.harness.pools import (
|
||
GlobalPoolStrategy,
|
||
PerCategoryPoolStrategy,
|
||
build_or_load_pools,
|
||
build_pools,
|
||
load_pools,
|
||
save_pools,
|
||
)
|
||
from core.types import GeneratedQuestion, PoolConfig
|
||
|
||
if TYPE_CHECKING:
|
||
from pathlib import Path
|
||
|
||
|
||
def _make_question(qid: str, task_type: str = "Action Reasoning") -> GeneratedQuestion:
|
||
"""构造测试用 GeneratedQuestion。
|
||
|
||
参数:
|
||
qid: 题目 ID。
|
||
task_type: 题型。
|
||
|
||
返回:
|
||
GeneratedQuestion 实例。
|
||
"""
|
||
return GeneratedQuestion(
|
||
question_id=qid,
|
||
video_id="video_001",
|
||
task_type=task_type,
|
||
question=f"Question {qid}?",
|
||
options=("A. opt1", "B. opt2", "C. opt3", "D. opt4"),
|
||
answer="A",
|
||
source_nodes=("node_1",),
|
||
difficulty="medium",
|
||
)
|
||
|
||
|
||
def _make_question_set(
|
||
n: int,
|
||
task_types: list[str] | None = None,
|
||
) -> list[GeneratedQuestion]:
|
||
"""构造 n 道题，交替分配题型。
|
||
|
||
参数:
|
||
n: 题目数量。
|
||
task_types: 可选题型列表，轮转分配；None 默认 2 类。
|
||
|
||
返回:
|
||
题目列表。
|
||
"""
|
||
types = task_types or ["Action Reasoning", "Information Synopsis"]
|
||
return [_make_question(f"q_{i:04d}", types[i % len(types)]) for i in range(n)]
|
||
|
||
|
||
def _make_correctness(
|
||
questions: list[GeneratedQuestion],
|
||
correct_ratio: float = 0.5,
|
||
) -> dict[str, bool]:
|
||
"""构造 correctness 字典，前 correct_ratio 比例标对。
|
||
|
||
参数:
|
||
questions: 题目列表。
|
||
correct_ratio: 对题占比。
|
||
|
||
返回:
|
||
question_id -> bool。
|
||
"""
|
||
n_correct = round(len(questions) * correct_ratio)
|
||
return {q.question_id: (i < n_correct) for i, q in enumerate(questions)}
|
||
|
||
|
||
class TestBuildPoolsMutualExclusion:
|
||
"""三池 question_id 互斥验证。"""
|
||
|
||
def test_build_pools_mutual_exclusion(self) -> None:
|
||
"""三池切分后，任意两池不共享 question_id。"""
|
||
questions = _make_question_set(200)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
|
||
pools = build_pools(
|
||
questions,
|
||
correctness,
|
||
diag_cfg={
|
||
"size": 30,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
val_cfg={
|
||
"size": 30,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
test_cfg={"size": 30},
|
||
baseline_run_id="run_baseline",
|
||
)
|
||
|
||
diag_ids = {q.question_id for q in pools.diagnosis}
|
||
val_ids = {q.question_id for q in pools.validation}
|
||
test_ids = {q.question_id for q in pools.test}
|
||
|
||
assert diag_ids & val_ids == set(), "诊断池与验证池有重叠"
|
||
assert diag_ids & test_ids == set(), "诊断池与测试池有重叠"
|
||
assert val_ids & test_ids == set(), "验证池与测试池有重叠"
|
||
|
||
assert len(diag_ids) == 30
|
||
assert len(val_ids) == 30
|
||
assert len(test_ids) == 30
|
||
|
||
|
||
class TestBuildPoolsTestNaturalDistribution:
|
||
"""test 池使用自然分布（correct_ratio=None）。"""
|
||
|
||
def test_build_pools_test_natural_distribution(self) -> None:
|
||
"""test 池不强制对错比例，保留候选池的自然分布。
|
||
|
||
构造 correctness 为 50% 对/50% 错，diag/val 用 correct_ratio=0.3
|
||
强制裁剪，test 池走自然分布（correct_ratio=None）。验证 test 池
|
||
不受 correct_ratio 约束。
|
||
"""
|
||
questions = _make_question_set(300)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
|
||
pools = build_pools(
|
||
questions,
|
||
correctness,
|
||
diag_cfg={
|
||
"size": 20,
|
||
"correct_ratio": 0.3,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
val_cfg={
|
||
"size": 20,
|
||
"correct_ratio": 0.3,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
test_cfg={"size": 20},
|
||
baseline_run_id="run_baseline",
|
||
)
|
||
|
||
# diag/val 被 correct_ratio=0.3 裁剪：round(20*0.3) = 6 对, 14 错
|
||
diag_correct = sum(1 for q in pools.diagnosis if correctness[q.question_id])
|
||
val_correct = sum(1 for q in pools.validation if correctness[q.question_id])
|
||
assert diag_correct == 6, "诊断池应强制 30% 对题"
|
||
assert val_correct == 6, "验证池应强制 30% 对题"
|
||
|
||
# test 池自然分布：不受 correct_ratio 约束
|
||
assert len(pools.test) == 20
|
||
|
||
|
||
class TestSaveLoadPoolsRoundtrip:
|
||
"""save/load 往返一致验证。"""
|
||
|
||
def test_save_load_pools_roundtrip(self, tmp_path: Path) -> None:
|
||
"""save_pools → load_pools 后全字段一致。"""
|
||
questions = _make_question_set(100)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
|
||
original = build_pools(
|
||
questions,
|
||
correctness,
|
||
diag_cfg={
|
||
"size": 15,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
val_cfg={
|
||
"size": 15,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
test_cfg={"size": 15},
|
||
baseline_run_id="run_001",
|
||
)
|
||
|
||
pools_path = tmp_path / "pools.json"
|
||
save_pools(original, pools_path)
|
||
restored = load_pools(pools_path)
|
||
|
||
# 标量字段
|
||
assert restored.baseline_run_id == original.baseline_run_id
|
||
assert restored.baseline_val_accuracy == pytest.approx(original.baseline_val_accuracy)
|
||
assert restored.correctness == original.correctness
|
||
|
||
# 三池逐题比对
|
||
for pool_name in ("diagnosis", "validation", "test"):
|
||
orig_list = getattr(original, pool_name)
|
||
rest_list = getattr(restored, pool_name)
|
||
assert len(rest_list) == len(orig_list), f"{pool_name} 长度不一致"
|
||
for o, r in zip(orig_list, rest_list, strict=False):
|
||
assert o.question_id == r.question_id
|
||
assert o.video_id == r.video_id
|
||
assert o.task_type == r.task_type
|
||
assert o.question == r.question
|
||
assert o.options == r.options
|
||
assert o.answer == r.answer
|
||
assert o.source_nodes == r.source_nodes
|
||
assert o.difficulty == r.difficulty
|
||
|
||
|
||
class TestLoadPoolsOldFormatReject:
|
||
"""旧格式 pools.json（无 test 键）→ ValueError。"""
|
||
|
||
def test_load_pools_old_format_reject(self, tmp_path: Path) -> None:
|
||
"""缺少 test 键的 pools.json 必须抛出 ValueError。"""
|
||
old_format = {
|
||
"baseline_run_id": "run_old",
|
||
"baseline_val_accuracy": 0.5,
|
||
"correctness": {},
|
||
"diagnosis": [],
|
||
"validation": [],
|
||
}
|
||
pools_path = tmp_path / "pools.json"
|
||
pools_path.write_text(json.dumps(old_format), encoding="utf-8")
|
||
|
||
with pytest.raises(ValueError, match="旧格式"):
|
||
load_pools(pools_path)
|
||
|
||
|
||
class TestBuildOrLoadPoolsFrozen:
|
||
"""build_or_load_pools 冻结复用：pools.json 存在时原样加载不重切。"""
|
||
|
||
def test_build_or_load_pools_frozen(self, tmp_path: Path) -> None:
|
||
"""pools.json 已存在时，build_or_load_pools 返回冻结内容。"""
|
||
questions = _make_question_set(60)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
|
||
frozen = build_pools(
|
||
questions,
|
||
correctness,
|
||
diag_cfg={
|
||
"size": 10,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
val_cfg={
|
||
"size": 10,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
test_cfg={"size": 10},
|
||
baseline_run_id="run_frozen",
|
||
)
|
||
|
||
pools_path = tmp_path / "pools.json"
|
||
save_pools(frozen, pools_path)
|
||
|
||
# build_or_load_pools 中 pools.json 存在 → 直接 load_pools
|
||
# 此处直接测试 load_pools 行为等价
|
||
loaded = load_pools(pools_path)
|
||
|
||
assert loaded.baseline_run_id == frozen.baseline_run_id
|
||
assert loaded.baseline_val_accuracy == pytest.approx(frozen.baseline_val_accuracy)
|
||
assert len(loaded.test) == len(frozen.test)
|
||
assert len(loaded.validation) == len(frozen.validation)
|
||
assert len(loaded.diagnosis) == len(frozen.diagnosis)
|
||
|
||
# question_id 完全一致
|
||
for pool_name in ("diagnosis", "validation", "test"):
|
||
orig_ids = [q.question_id for q in getattr(frozen, pool_name)]
|
||
load_ids = [q.question_id for q in getattr(loaded, pool_name)]
|
||
assert orig_ids == load_ids, f"{pool_name} 冻结后 ID 顺序不一致"
|
||
|
||
def _run_config_for_frozen(self, tmp_path: Path, seed_name: str) -> object:
|
||
"""构造指向 tmp workspace/store + 指定种子名的最小 train RunConfig。"""
|
||
from app.harness.config import RunConfig
|
||
|
||
return RunConfig(
|
||
workspace_dir=tmp_path / "ws",
|
||
store_dir=tmp_path / "store",
|
||
mode="train",
|
||
concurrency=4,
|
||
max_steps=10,
|
||
skill_mode="auto",
|
||
n_samples=0,
|
||
questions="benchmarks/Video-MME",
|
||
skills_version="v1",
|
||
prompts_version="v1",
|
||
epochs=1,
|
||
diag_size=10,
|
||
diag_correct_ratio=0.5,
|
||
val_size=10,
|
||
val_correct_ratio=0.5,
|
||
edit_budget_start=5,
|
||
edit_budget_end=2,
|
||
batch_size=15,
|
||
min_class_per_batch=2,
|
||
eval_min_per_class=2,
|
||
early_stop_patience=4,
|
||
test_size=10,
|
||
use_slow_momentum=True,
|
||
gate_e_confirm=20.0,
|
||
gate_e_provisional=3.0,
|
||
gate_w_net_min=2,
|
||
gate_delta_min=0.02,
|
||
gate_lambda_dir=-0.642,
|
||
gate_e_rollback=10.0,
|
||
gate_block=8,
|
||
gate_n_max=40,
|
||
gate_p_low=0.05,
|
||
gate_p_high=0.95,
|
||
gate_probe_quota=0.2,
|
||
gate_gamma_decay=0.9,
|
||
gate_cooldown_steps=2,
|
||
gate_guard_err=0.10,
|
||
skill_update_mode="patch",
|
||
appendix_consolidate_threshold=6,
|
||
fresh=True,
|
||
seed=seed_name,
|
||
test_questions="", # 绕过 _to_pool_config 的 resolve_paths(manifest 依赖)
|
||
)
|
||
|
||
def test_global_frozen_rejects_baseline_mismatch(self, tmp_path: Path) -> None:
|
||
"""global 冻结 pools 的 baseline_run_id 与 seed 不符时 fail-loud。"""
|
||
seed_name = "myseed"
|
||
seed_dir = tmp_path / "store" / "seeds" / seed_name
|
||
seed_dir.mkdir(parents=True)
|
||
(seed_dir / "seed.json").write_text(
|
||
json.dumps({"baseline_run_id": "infer_adhoc", "parent": None})
|
||
)
|
||
|
||
ws = tmp_path / "ws"
|
||
ws.mkdir()
|
||
(ws / "pools.json").write_text(
|
||
json.dumps({"split_mode": "global", "baseline_run_id": "other"}),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
config = self._run_config_for_frozen(tmp_path, seed_name)
|
||
strategy = GlobalPoolStrategy()
|
||
with pytest.raises(ValueError, match="baseline_run_id"):
|
||
build_or_load_pools(config, strategy, tmp_path / "nonexistent.db")
|
||
|
||
|
||
class TestGlobalPoolStrategy:
|
||
"""GlobalPoolStrategy 封装现有全局三分逻辑。"""
|
||
|
||
def test_global_strategy_builds_three_pools(self) -> None:
|
||
"""GlobalPoolStrategy.build 产出三个互斥池。"""
|
||
questions = _make_question_set(200)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=42,
|
||
baseline_run_id="run_baseline",
|
||
diag_size=30,
|
||
diag_correct_ratio=0.5,
|
||
val_size=30,
|
||
val_correct_ratio=0.5,
|
||
test_size=30,
|
||
eval_min_per_class=1,
|
||
train_ratio=0.667,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = GlobalPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
diag_ids = {q.question_id for q in pools.diagnosis}
|
||
val_ids = {q.question_id for q in pools.validation}
|
||
test_ids = {q.question_id for q in pools.test}
|
||
assert diag_ids & val_ids == set()
|
||
assert diag_ids & test_ids == set()
|
||
assert val_ids & test_ids == set()
|
||
assert len(pools.diagnosis) == 30
|
||
assert len(pools.validation) == 30
|
||
assert len(pools.test) == 30
|
||
|
||
def test_global_strategy_build_incremental_raises(self) -> None:
|
||
"""GlobalPoolStrategy 不支持增量。"""
|
||
strategy = GlobalPoolStrategy()
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=0,
|
||
baseline_run_id="r",
|
||
diag_size=10,
|
||
diag_correct_ratio=0.5,
|
||
val_size=10,
|
||
val_correct_ratio=0.5,
|
||
test_size=10,
|
||
eval_min_per_class=1,
|
||
train_ratio=0.667,
|
||
test_questions_dir=None,
|
||
)
|
||
with pytest.raises(NotImplementedError):
|
||
strategy.build_incremental(["Action Reasoning"], [], {}, config)
|
||
|
||
|
||
def _make_per_category_questions():
|
||
"""构造 12 类各 30 题，共 360 题。"""
|
||
task_types = [
|
||
"Action Recognition",
|
||
"Action Reasoning",
|
||
"Attribute Perception",
|
||
"Counting Problem",
|
||
"Information Synopsis",
|
||
"Object Recognition",
|
||
"Object Reasoning",
|
||
"OCR Problems",
|
||
"Spatial Perception",
|
||
"Spatial Reasoning",
|
||
"Temporal Perception",
|
||
"Temporal Reasoning",
|
||
]
|
||
questions = []
|
||
for tt in task_types:
|
||
for i in range(30):
|
||
questions.append(_make_question(f"{tt}_{i:03d}", tt))
|
||
return questions
|
||
|
||
|
||
class TestPerCategoryPoolStrategy:
|
||
"""PerCategoryPoolStrategy per-category 2:1 分层划分。"""
|
||
|
||
def test_per_category_split_20_10(self):
|
||
"""每类 30 题按 correctness 2:1 分层 -> 20 train + 10 val。"""
|
||
questions = _make_per_category_questions()
|
||
correctness = {}
|
||
for q in questions:
|
||
idx = int(q.question_id.split("_")[-1])
|
||
correctness[q.question_id] = idx < 18
|
||
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=42,
|
||
baseline_run_id="baseline_v2",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
assert len(pools.diagnosis) == 240
|
||
assert len(pools.validation) == 120
|
||
|
||
from collections import Counter
|
||
|
||
diag_counts = Counter(q.task_type for q in pools.diagnosis)
|
||
val_counts = Counter(q.task_type for q in pools.validation)
|
||
for tt in diag_counts:
|
||
assert diag_counts[tt] == 20
|
||
assert val_counts[tt] == 10
|
||
|
||
diag_ids = {q.question_id for q in pools.diagnosis}
|
||
val_ids = {q.question_id for q in pools.validation}
|
||
assert diag_ids & val_ids == set()
|
||
|
||
def test_per_category_correctness_ratio_aligned(self):
|
||
"""train 和 val 的 correctness 比例应对齐。"""
|
||
questions = _make_per_category_questions()
|
||
correctness = {}
|
||
for q in questions:
|
||
idx = int(q.question_id.split("_")[-1])
|
||
correctness[q.question_id] = idx < 18
|
||
|
||
config = PoolConfig(
|
||
task_types=("Action Reasoning",),
|
||
seed=42,
|
||
baseline_run_id="baseline_v2",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
assert len(pools.diagnosis) == 20
|
||
assert len(pools.validation) == 10
|
||
diag_correct = sum(1 for q in pools.diagnosis if correctness[q.question_id])
|
||
val_correct = sum(1 for q in pools.validation if correctness[q.question_id])
|
||
assert diag_correct == 12
|
||
assert val_correct == 6
|
||
|
||
def test_per_category_all_correct_degrades(self):
|
||
"""某类全部 correct -> 退化为非分层 random 20/10。"""
|
||
questions = [_make_question(f"q_{i:03d}", "Object Recognition") for i in range(30)]
|
||
correctness = {q.question_id: True for q in questions}
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=42,
|
||
baseline_run_id="r",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
assert len(pools.diagnosis) == 20
|
||
assert len(pools.validation) == 10
|
||
|
||
def test_per_category_missing_correctness_fails(self):
|
||
"""correctness 不完整时 fail-fast。"""
|
||
questions = [_make_question(f"q_{i:03d}", "Object Recognition") for i in range(30)]
|
||
correctness = {q.question_id: True for q in questions[:25]}
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=42,
|
||
baseline_run_id="r",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
with pytest.raises(ValueError, match="correctness 缺失"):
|
||
strategy.build(questions, correctness, config)
|
||
|
||
def test_per_category_task_types_filter(self):
|
||
"""task_types 过滤只处理指定类别。"""
|
||
questions = _make_per_category_questions()
|
||
correctness = {q.question_id: True for q in questions}
|
||
config = PoolConfig(
|
||
task_types=("Action Reasoning", "Information Synopsis"),
|
||
seed=42,
|
||
baseline_run_id="r",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
assert len(pools.diagnosis) == 40
|
||
assert len(pools.validation) == 20
|
||
types_in_diag = {q.task_type for q in pools.diagnosis}
|
||
assert types_in_diag == {"Action Reasoning", "Information Synopsis"}
|
||
|
||
def test_per_category_test_pool_filtered_by_task_types(self, tmp_path: Path) -> None:
|
||
"""test_questions_dir 加载的 test 池应按 task_types 过滤。"""
|
||
|
||
test_dir = tmp_path / "test_questions"
|
||
test_dir.mkdir()
|
||
|
||
task_types_all = [
|
||
"Action Recognition",
|
||
"Action Reasoning",
|
||
"Temporal Perception",
|
||
]
|
||
for tt in task_types_all:
|
||
items = []
|
||
for i in range(10):
|
||
items.append(
|
||
{
|
||
"question_id": f"{tt}_{i:03d}",
|
||
"video_id": "v1",
|
||
"task_type": tt,
|
||
"question": f"Q {tt} {i}?",
|
||
"options": ["A. a", "B. b", "C. c", "D. d"],
|
||
"answer": "A",
|
||
}
|
||
)
|
||
slug = tt.lower().replace(" ", "_")
|
||
(test_dir / f"{slug}.json").write_text(json.dumps(items, ensure_ascii=False))
|
||
|
||
# train/val 用的题目（与 test 独立）
|
||
questions = _make_per_category_questions()
|
||
correctness = {q.question_id: True for q in questions}
|
||
|
||
config = PoolConfig(
|
||
task_types=("Action Recognition",),
|
||
seed=42,
|
||
baseline_run_id="baseline_v2",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=test_dir,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
|
||
assert len(pools.test) == 10, (
|
||
f"test 池应仅含 Action Recognition 的 10 题，实际 {len(pools.test)}"
|
||
)
|
||
test_types = {q.task_type for q in pools.test}
|
||
assert test_types == {"Action Recognition"}, (
|
||
f"test 池应仅含 Action Recognition，实际含 {test_types}"
|
||
)
|
||
|
||
def test_per_category_test_pool_no_filter_when_task_types_none(self, tmp_path: Path) -> None:
|
||
"""task_types=None 时，test 池不过滤，加载全部题目。"""
|
||
|
||
test_dir = tmp_path / "test_questions"
|
||
test_dir.mkdir()
|
||
|
||
task_types_all = [
|
||
"Action Recognition",
|
||
"Action Reasoning",
|
||
"Temporal Perception",
|
||
]
|
||
total_expected = 0
|
||
for tt in task_types_all:
|
||
items = []
|
||
for i in range(10):
|
||
items.append(
|
||
{
|
||
"question_id": f"{tt}_{i:03d}",
|
||
"video_id": "v1",
|
||
"task_type": tt,
|
||
"question": f"Q {tt} {i}?",
|
||
"options": ["A. a", "B. b", "C. c", "D. d"],
|
||
"answer": "A",
|
||
}
|
||
)
|
||
slug = tt.lower().replace(" ", "_")
|
||
(test_dir / f"{slug}.json").write_text(json.dumps(items, ensure_ascii=False))
|
||
total_expected += len(items)
|
||
|
||
questions = _make_per_category_questions()
|
||
correctness = {q.question_id: True for q in questions}
|
||
|
||
config = PoolConfig(
|
||
task_types=None,
|
||
seed=42,
|
||
baseline_run_id="baseline_v2",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=test_dir,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
|
||
assert len(pools.test) == total_expected, (
|
||
f"task_types=None 时应加载全部 {total_expected} 题，实际 {len(pools.test)}"
|
||
)
|
||
|
||
|
||
class TestPerCategorySaveLoad:
|
||
"""per_category 格式的 pools.json 冻结/加载。"""
|
||
|
||
def test_save_load_per_category_roundtrip(self, tmp_path: Path) -> None:
|
||
"""per_category 模式 save -> load 往返一致。"""
|
||
questions = _make_per_category_questions()
|
||
correctness = {q.question_id: True for q in questions}
|
||
config = PoolConfig(
|
||
task_types=("Action Reasoning",),
|
||
seed=42,
|
||
baseline_run_id="baseline_v2",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
pools_path = tmp_path / "pools.json"
|
||
save_pools(pools, pools_path, split_mode="per_category", config=config)
|
||
loaded = load_pools(pools_path)
|
||
assert loaded.baseline_run_id == pools.baseline_run_id
|
||
assert len(loaded.diagnosis) == len(pools.diagnosis)
|
||
assert len(loaded.validation) == len(pools.validation)
|
||
|
||
# 验证 per_category 格式内容
|
||
data = json.loads(pools_path.read_text())
|
||
assert data["split_mode"] == "per_category"
|
||
assert "categories" in data
|
||
assert data["seed"] == 42
|
||
assert data["train_ratio"] == pytest.approx(20 / 30)
|
||
assert data["test_source"] is None
|
||
|
||
# categories 内容校验
|
||
cats = data["categories"]
|
||
assert "Action Reasoning" in cats
|
||
assert len(cats["Action Reasoning"]["train"]) == 20
|
||
assert len(cats["Action Reasoning"]["val"]) == 10
|
||
|
||
def test_save_per_category_without_config_raises(self, tmp_path: Path) -> None:
|
||
"""per_category 模式未提供 config 时报 ValueError。"""
|
||
from app.harness.pools import Pools
|
||
|
||
pools = Pools(
|
||
diagnosis=[],
|
||
validation=[],
|
||
test=[],
|
||
baseline_run_id="r",
|
||
baseline_val_accuracy=0.0,
|
||
)
|
||
with pytest.raises(ValueError, match="per_category 模式下.*必须提供 config"):
|
||
save_pools(pools, tmp_path / "pools.json", split_mode="per_category")
|
||
|
||
def test_save_global_mode_has_split_mode_field(self, tmp_path: Path) -> None:
|
||
"""global 模式 save 也写入 split_mode 字段。"""
|
||
questions = _make_question_set(60)
|
||
correctness = _make_correctness(questions, 0.5)
|
||
original = build_pools(
|
||
questions,
|
||
correctness,
|
||
diag_cfg={
|
||
"size": 10,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
val_cfg={
|
||
"size": 10,
|
||
"correct_ratio": 0.5,
|
||
"task_types": None,
|
||
"seed": 42,
|
||
"min_per_class": None,
|
||
},
|
||
test_cfg={"size": 10},
|
||
baseline_run_id="run_001",
|
||
)
|
||
pools_path = tmp_path / "pools.json"
|
||
save_pools(original, pools_path, split_mode="global")
|
||
data = json.loads(pools_path.read_text())
|
||
assert data["split_mode"] == "global"
|
||
assert "categories" not in data
|
||
|
||
# 仍能正常 load
|
||
loaded = load_pools(pools_path)
|
||
assert loaded.baseline_run_id == "run_001"
|
||
assert len(loaded.diagnosis) == 10
|
||
|
||
def test_load_legacy_format_without_split_mode(self, tmp_path: Path) -> None:
|
||
"""旧格式（无 split_mode 字段）仍可加载。"""
|
||
legacy = {
|
||
"baseline_run_id": "run_legacy",
|
||
"baseline_val_accuracy": 0.75,
|
||
"correctness": {"q1": True},
|
||
"diagnosis": [
|
||
{
|
||
"question_id": "q1",
|
||
"video_id": "v1",
|
||
"task_type": "AR",
|
||
"question": "Q?",
|
||
"options": ["A", "B", "C", "D"],
|
||
"answer": "A",
|
||
"source_nodes": [],
|
||
"difficulty": "medium",
|
||
"skill_target": None,
|
||
"difficulty_steps": None,
|
||
}
|
||
],
|
||
"validation": [],
|
||
"test": [],
|
||
}
|
||
pools_path = tmp_path / "pools.json"
|
||
pools_path.write_text(json.dumps(legacy), encoding="utf-8")
|
||
loaded = load_pools(pools_path)
|
||
assert loaded.baseline_run_id == "run_legacy"
|
||
assert len(loaded.diagnosis) == 1
|
||
|
||
def test_per_category_categories_multi_type(self, tmp_path: Path) -> None:
|
||
"""多类别 per_category save 后 categories 包含所有类别。"""
|
||
questions = _make_per_category_questions()
|
||
correctness = {q.question_id: True for q in questions}
|
||
config = PoolConfig(
|
||
task_types=("Action Reasoning", "Information Synopsis"),
|
||
seed=0,
|
||
baseline_run_id="b",
|
||
diag_size=0,
|
||
diag_correct_ratio=0.0,
|
||
val_size=0,
|
||
val_correct_ratio=0.0,
|
||
test_size=0,
|
||
eval_min_per_class=0,
|
||
train_ratio=20 / 30,
|
||
test_questions_dir=None,
|
||
)
|
||
strategy = PerCategoryPoolStrategy()
|
||
pools = strategy.build(questions, correctness, config)
|
||
pools_path = tmp_path / "pools.json"
|
||
save_pools(pools, pools_path, split_mode="per_category", config=config)
|
||
|
||
data = json.loads(pools_path.read_text())
|
||
assert set(data["categories"].keys()) == {
|
||
"Action Reasoning",
|
||
"Information Synopsis",
|
||
}
|
||
for tt in data["categories"]:
|
||
cat = data["categories"][tt]
|
||
assert len(cat["train"]) == 20
|
||
assert len(cat["val"]) == 10
|
||
# train + val 的 qid 互斥
|
||
assert set(cat["train"]) & set(cat["val"]) == set()
|
||
|
||
|
||
class TestRunHoldoutEvalConfig:
|
||
"""run_holdout_eval 字段校验。"""
|
||
|
||
def test_default_true(self):
|
||
"""run_holdout_eval 默认值为 True。"""
|
||
from pathlib import Path
|
||
|
||
from app.harness.config import RunConfig
|
||
|
||
config = RunConfig(
|
||
workspace_dir=Path("/tmp/ws"),
|
||
store_dir=Path("/tmp/store"),
|
||
mode="train",
|
||
concurrency=4,
|
||
max_steps=10,
|
||
skill_mode="auto",
|
||
n_samples=0,
|
||
questions="benchmarks/Video-MME",
|
||
skills_version="v1",
|
||
prompts_version="v1",
|
||
epochs=1,
|
||
diag_size=100,
|
||
diag_correct_ratio=0.5,
|
||
val_size=30,
|
||
val_correct_ratio=0.5,
|
||
edit_budget_start=5,
|
||
edit_budget_end=2,
|
||
batch_size=15,
|
||
min_class_per_batch=2,
|
||
eval_min_per_class=2,
|
||
early_stop_patience=4,
|
||
test_size=30,
|
||
use_slow_momentum=True,
|
||
gate_e_confirm=20.0,
|
||
gate_e_provisional=3.0,
|
||
gate_w_net_min=2,
|
||
gate_delta_min=0.02,
|
||
gate_lambda_dir=-0.642,
|
||
gate_e_rollback=10.0,
|
||
gate_block=8,
|
||
gate_n_max=40,
|
||
gate_p_low=0.05,
|
||
gate_p_high=0.95,
|
||
gate_probe_quota=0.2,
|
||
gate_gamma_decay=0.9,
|
||
gate_cooldown_steps=2,
|
||
gate_guard_err=0.10,
|
||
skill_update_mode="patch",
|
||
appendix_consolidate_threshold=6,
|
||
run_id="test_run",
|
||
)
|
||
assert config.run_holdout_eval is True
|
||
|
||
def test_explicit_false(self):
|
||
"""run_holdout_eval 可设为 False。"""
|
||
from pathlib import Path
|
||
|
||
from app.harness.config import RunConfig
|
||
|
||
config = RunConfig(
|
||
workspace_dir=Path("/tmp/ws"),
|
||
store_dir=Path("/tmp/store"),
|
||
mode="train",
|
||
concurrency=4,
|
||
max_steps=10,
|
||
skill_mode="auto",
|
||
n_samples=0,
|
||
questions="benchmarks/Video-MME",
|
||
skills_version="v1",
|
||
prompts_version="v1",
|
||
epochs=1,
|
||
diag_size=100,
|
||
diag_correct_ratio=0.5,
|
||
val_size=30,
|
||
val_correct_ratio=0.5,
|
||
edit_budget_start=5,
|
||
edit_budget_end=2,
|
||
batch_size=15,
|
||
min_class_per_batch=2,
|
||
eval_min_per_class=2,
|
||
early_stop_patience=4,
|
||
test_size=30,
|
||
use_slow_momentum=True,
|
||
gate_e_confirm=20.0,
|
||
gate_e_provisional=3.0,
|
||
gate_w_net_min=2,
|
||
gate_delta_min=0.02,
|
||
gate_lambda_dir=-0.642,
|
||
gate_e_rollback=10.0,
|
||
gate_block=8,
|
||
gate_n_max=40,
|
||
gate_p_low=0.05,
|
||
gate_p_high=0.95,
|
||
gate_probe_quota=0.2,
|
||
gate_gamma_decay=0.9,
|
||
gate_cooldown_steps=2,
|
||
gate_guard_err=0.10,
|
||
skill_update_mode="patch",
|
||
appendix_consolidate_threshold=6,
|
||
run_id="test_run",
|
||
run_holdout_eval=False,
|
||
)
|
||
assert config.run_holdout_eval is False
|