"""视频级切分科研旋钮单元测试:诊断指纹 + val_wrong_min 功效护栏 + PoolConfig 新字段。 覆盖: - diag_fingerprint 对 (prompt 版本 / 模型 / 代码版本) 三元组确定且敏感; - split_by_video_assignment 的 val_wrong_min 门控 fail loud(验证信号不足即报错); - val_wrong_min 默认 0 时行为与 Task 11 现有调用完全一致(不回归); - PoolConfig 能接收视频级切分的五个新旋钮字段(纯 dataclass 装配)。 """ from __future__ import annotations import pytest from app.harness.pools import InsufficientValSignal, split_by_video_assignment from app.harness.split_selection import diag_fingerprint from core.types import GeneratedQuestion, PoolConfig def _q(qid: str, vid: str, tt: str = "Counting Problem") -> GeneratedQuestion: """构造最小可用题目(补齐 GeneratedQuestion 的必填 source_nodes/difficulty)。""" return GeneratedQuestion( question_id=qid, video_id=vid, task_type=tt, question="", options=("A", "B", "C", "D"), answer="A", source_nodes=(), difficulty="easy", ) def test_diag_fingerprint_deterministic_and_sensitive(): """诊断指纹对相同三元组稳定,对任一分量变化敏感。""" a = diag_fingerprint("p1", "deepseek-v4", "abc123") assert a == diag_fingerprint("p1", "deepseek-v4", "abc123") # 确定性 assert a != diag_fingerprint("p2", "deepseek-v4", "abc123") # prompt 变则变 assert a != diag_fingerprint("p1", "kimi", "abc123") # model 变则变 assert a != diag_fingerprint("p1", "deepseek-v4", "def456") # 代码版本变则变 assert len(a) == 16 # sha256 截断 16 位十六进制 def test_val_wrong_min_enforced(): """val 错题数 < val_wrong_min 时 fail loud(InsufficientValSignal),不静默兜底。""" qs = [_q("v1-1", "v1"), _q("v1-2", "v1"), _q("v1-3", "v1")] correctness = {q.question_id: True for q in qs} # 全对 → val 无错题 with pytest.raises(InsufficientValSignal): # val 错题 < val_wrong_min split_by_video_assignment( qs, {"v1": "trainval"}, correctness=correctness, val_ratio=0.5, seed=0, val_wrong_min=5, ) def test_val_wrong_min_default_zero_no_regression(): """val_wrong_min 默认 0 时不检查错题数,保持 Task 11 现有调用契约不破。""" qs = [_q("v1-1", "v1"), _q("v2-1", "v2")] assignment = {"v1": "trainval", "v2": "trainval"} correctness = {"v1-1": True, "v2-1": True} # 全对但默认不触发护栏 pools = split_by_video_assignment( qs, assignment, correctness=correctness, val_ratio=1.0, seed=0 ) assert len(pools.validation) == 2 # 未抛异常,正常返回 def test_pool_config_accepts_video_split_knobs(): """PoolConfig 能接收视频级切分五个新旋钮字段(默认惰性,不破坏现有构造点)。""" cfg = PoolConfig( task_types=None, seed=0, baseline_run_id="infer_adhoc", diag_size=200, diag_correct_ratio=0.5, val_size=30, val_correct_ratio=0.5, test_size=60, eval_min_per_class=2, train_ratio=0.667, test_questions_dir=None, n_trainval=100, floor_k={"Counting Problem": 3}, epsilon=0.1, report_floor=27, val_wrong_min=20, ) assert cfg.n_trainval == 100 assert cfg.floor_k == {"Counting Problem": 3} assert cfg.epsilon == 0.1 assert cfg.report_floor == 27 assert cfg.val_wrong_min == 20 def test_pool_config_video_split_knobs_default_inert(): """未传视频级切分字段时默认惰性(0 / 空 dict),不破坏 GlobalPoolStrategy 现有构造。""" cfg = PoolConfig( task_types=None, seed=0, baseline_run_id="run_1", diag_size=200, diag_correct_ratio=0.5, val_size=30, val_correct_ratio=0.5, test_size=60, eval_min_per_class=2, train_ratio=0.667, test_questions_dir=None, ) assert cfg.n_trainval == 0 assert cfg.floor_k == {} assert cfg.epsilon == 0.0 assert cfg.report_floor == 0 assert cfg.val_wrong_min == 0