8038109ce8
run_pair_dual_sync:同一份 staged src/ 上先自动 spec(ensure_play_spec)、后金标 (inject_golden)各 play 一次,取两组逐门 verdict——隔离生成方差、只暴露驱动器差。 编排序 gen-only→smoke→断言 staged 无残留 spec→ensure→play(auto)→inject→play (golden);auto_gates 在金标注入前捕获。run_batch 端口池+线程前台有界并发(≤15 夹取、复用 bake_off._clamp_conc)、报告 auto-vs-golden-<idx>.json 不覆盖。 mock 编排单测(调用序 + auto 先于 golden + no-spec 守卫)20/20 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
286 lines
13 KiB
Python
286 lines
13 KiB
Python
"""
|
|
test_auto_vs_golden.py — M2 auto-vs-golden 门单测(U1 逐门 delta 判据 + U2 编排 mock + U3 退役授权)。
|
|
|
|
守的不变量(U1):
|
|
· 逐门 delta = auto 过门率 - golden 过门率;关键门 {E_live,H_progress,G_input} 容差 0(delta≥0)、
|
|
其余门容差 -1/N(允许单款 flake)。
|
|
· 方向单边:查「自动不比金标驱得差」(delta<0 退化),非查「自动比金标松」。
|
|
· double-low 守卫:关键门 auto 与 golden 过门率都 <0.5(两路同挂)→ inconclusive(红)、不误判 aligned。
|
|
· 按品类聚合:三品类全 aligned 才整体 meets;任一 regressed/inconclusive/缺样本 → meets=False。
|
|
· 零 LLM:纯函数、同输入恒同输出。
|
|
|
|
跑:cheap-worker/.venv/bin/python cheap-worker/tests/test_auto_vs_golden.py
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
|
import auto_vs_golden as A # noqa: E402
|
|
|
|
_ALL_GATES = ["A_boot", "B_uncaught", "C_frame", "D_render", "E_live",
|
|
"F_wiring", "G_input", "H_progress", "I_control"]
|
|
|
|
|
|
def _gates(fail=()):
|
|
"""九门全过,fail 里的门置 False。"""
|
|
return {g: (g not in fail) for g in _ALL_GATES}
|
|
|
|
|
|
def _run(auto_fail=(), golden_fail=()):
|
|
"""一款:auto/golden 两组逐门(默认全过,指定门挂)。"""
|
|
return {"autoGates": _gates(auto_fail), "goldenGates": _gates(golden_fail)}
|
|
|
|
|
|
# ───────────────────────── U1 逐门 delta 判据 ─────────────────────────
|
|
|
|
def test_delta_all_equal_aligned():
|
|
"""auto 每门 = golden(delta 全 0)→ aligned、meets。"""
|
|
j = A.judge_genre_delta([_run() for _ in range(5)])
|
|
assert j["status"] == "aligned" and j["meets"] is True
|
|
assert j["perGate"]["H_progress"]["delta"] == 0.0
|
|
|
|
|
|
def test_delta_key_gate_lower_one_sample_regressed():
|
|
"""关键门 H_progress auto 比 golden 低一款(delta=-0.2)→ regressed(关键门容差 0)、meets=False。"""
|
|
runs = [_run(auto_fail=("H_progress",))] + [_run() for _ in range(4)]
|
|
j = A.judge_genre_delta(runs)
|
|
assert j["status"] == "regressed" and j["meets"] is False
|
|
assert "H_progress" in j["regressedGates"]
|
|
assert j["perGate"]["H_progress"]["delta"] == -0.2
|
|
|
|
|
|
def test_delta_nonkey_gate_lower_one_sample_within_tolerance():
|
|
"""非关键门 D_render auto 低一款(delta=-1/5=-0.2)→ 在容差内 → aligned。"""
|
|
runs = [_run(auto_fail=("D_render",))] + [_run() for _ in range(4)]
|
|
j = A.judge_genre_delta(runs)
|
|
assert j["status"] == "aligned" and j["meets"] is True
|
|
|
|
|
|
def test_delta_nonkey_gate_lower_two_samples_regressed():
|
|
"""非关键门 D_render auto 低两款(delta=-0.4 < -1/5)→ regressed。"""
|
|
runs = [_run(auto_fail=("D_render",)), _run(auto_fail=("D_render",))] + [_run() for _ in range(3)]
|
|
j = A.judge_genre_delta(runs)
|
|
assert j["status"] == "regressed" and "D_render" in j["regressedGates"]
|
|
|
|
|
|
def test_delta_auto_higher_than_golden_not_regressed():
|
|
"""方向单边:auto 关键门比 golden 高(golden 挂、auto 过,delta=+0.2)→ 不算退化、aligned
|
|
(门查的是 auto 不比 golden 差,不查 auto 比 golden 松)。"""
|
|
runs = [_run(golden_fail=("H_progress",))] + [_run() for _ in range(4)]
|
|
j = A.judge_genre_delta(runs)
|
|
assert j["status"] == "aligned" and j["meets"] is True
|
|
assert j["perGate"]["H_progress"]["delta"] == 0.2
|
|
|
|
|
|
def test_delta_double_low_key_gate_inconclusive():
|
|
"""double-low:关键门 E_live auto 与 golden 过门率都 <0.5(5 款里 3 款两路同挂)→ inconclusive(红)、
|
|
meets=False,不被 delta=0 误判 aligned。"""
|
|
runs = [_run(auto_fail=("E_live",), golden_fail=("E_live",)) for _ in range(3)] + [_run() for _ in range(2)]
|
|
j = A.judge_genre_delta(runs)
|
|
assert j["status"] == "inconclusive" and j["meets"] is False
|
|
assert "E_live" in j["inconclusiveGates"]
|
|
|
|
|
|
def test_delta_insufficient_empty():
|
|
j = A.judge_genre_delta([])
|
|
assert j["status"] == "insufficient" and j["meets"] is False and j["total"] == 0
|
|
|
|
|
|
def test_delta_deterministic_zero_llm():
|
|
"""纯函数、同输入恒同输出。"""
|
|
runs = [_run(auto_fail=("D_render",))] + [_run() for _ in range(4)]
|
|
assert A.judge_genre_delta(runs) == A.judge_genre_delta(runs)
|
|
|
|
|
|
# ───────────────────────── U1 按品类聚合 ─────────────────────────
|
|
|
|
def test_aggregate_all_aligned_meets():
|
|
gr = {"click-score": [_run() for _ in range(5)],
|
|
"whack-mole": [_run() for _ in range(5)],
|
|
"shop-serve": [_run() for _ in range(5)]}
|
|
r = A.aggregate_delta(gr, ["click-score", "whack-mole", "shop-serve"])
|
|
assert r["meets"] is True and not r["regressedGenres"] and not r["missingGenres"]
|
|
|
|
|
|
def test_aggregate_one_regressed_fails():
|
|
"""某品类关键门退化 → 整体 meets=False、不被其余品类平均掩盖。"""
|
|
gr = {"click-score": [_run() for _ in range(5)],
|
|
"whack-mole": [_run(auto_fail=("H_progress",))] + [_run() for _ in range(4)],
|
|
"shop-serve": [_run() for _ in range(5)]}
|
|
r = A.aggregate_delta(gr, ["click-score", "whack-mole", "shop-serve"])
|
|
assert r["meets"] is False and "whack-mole" in r["regressedGenres"]
|
|
|
|
|
|
def test_aggregate_missing_genre_fails():
|
|
gr = {"click-score": [_run() for _ in range(5)], "whack-mole": [_run() for _ in range(5)]}
|
|
r = A.aggregate_delta(gr, ["click-score", "whack-mole", "shop-serve"])
|
|
assert r["meets"] is False and "shop-serve" in r["missingGenres"]
|
|
|
|
|
|
# ───────────────────────── U3 Node 退役授权三条齐判定 ─────────────────────────
|
|
|
|
_REQ = ["click-score", "whack-mole", "shop-serve"]
|
|
|
|
|
|
def _write(d: Path, name: str, obj) -> Path:
|
|
p = d / name
|
|
p.write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8")
|
|
return p
|
|
|
|
|
|
def _reports(d: Path, m1=True, equiv=None, avg=True):
|
|
"""造三份报告:M1 达标 / 002 等价品类集 / M2 auto-vs-golden meets。"""
|
|
m1p = _write(d, "m1.json", {"overallMeets": m1})
|
|
equiv = _REQ if equiv is None else equiv
|
|
cmp = _write(d, "cmp.json", {"summary": {"equivalentGenres": equiv}})
|
|
avgp = _write(d, "avg.json", {"meets": avg})
|
|
return m1p, cmp, avgp
|
|
|
|
|
|
def test_retire_all_three_green_authorized():
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t))
|
|
r = A.retire_authorization(m1p, cmp, avgp, _REQ)
|
|
assert r["authorized"] is True
|
|
assert r["conditions"] == {"m1Met": True, "parityEquivalent": True, "autoVsGoldenAligned": True}
|
|
assert not r["missing"]
|
|
|
|
|
|
def test_retire_m1_not_met_unauthorized():
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t), m1=False)
|
|
r = A.retire_authorization(m1p, cmp, avgp, _REQ)
|
|
assert r["authorized"] is False and r["conditions"]["m1Met"] is False
|
|
assert any("M1" in m or "达标" in m for m in r["missing"])
|
|
|
|
|
|
def test_retire_parity_missing_genre_unauthorized():
|
|
"""002 等价品类集缺一品类(某品类 regressed)→ 对照等价未达成。"""
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t), equiv=["click-score", "whack-mole"])
|
|
r = A.retire_authorization(m1p, cmp, avgp, _REQ)
|
|
assert r["authorized"] is False and r["conditions"]["parityEquivalent"] is False
|
|
assert any("对照" in m or "equivalent" in m.lower() for m in r["missing"])
|
|
|
|
|
|
def test_retire_avg_not_met_unauthorized():
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t), avg=False)
|
|
r = A.retire_authorization(m1p, cmp, avgp, _REQ)
|
|
assert r["authorized"] is False and r["conditions"]["autoVsGoldenAligned"] is False
|
|
assert any("auto-vs-golden" in m or "auto" in m.lower() for m in r["missing"])
|
|
|
|
|
|
def test_retire_missing_report_file_unauthorized():
|
|
"""任一报告文件缺失 → 前置证据缺失、不静默当通过。"""
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t))
|
|
r = A.retire_authorization(m1p, cmp, Path(t) / "nonexistent.json", _REQ)
|
|
assert r["authorized"] is False and r["conditions"]["autoVsGoldenAligned"] is False
|
|
assert any("缺失" in m or "证据" in m for m in r["missing"])
|
|
|
|
|
|
def test_retire_deterministic_zero_llm():
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m1p, cmp, avgp = _reports(Path(t))
|
|
assert A.retire_authorization(m1p, cmp, avgp, _REQ) == A.retire_authorization(m1p, cmp, avgp, _REQ)
|
|
|
|
|
|
# ───────────────────────── U2 同游戏双驱动 play 编排(mock · 不真跑模型)─────────────────────────
|
|
|
|
def _patch(obj, name, val, saved):
|
|
saved.append((obj, name, getattr(obj, name)))
|
|
setattr(obj, name, val)
|
|
|
|
|
|
def _restore(saved):
|
|
for obj, name, old in saved:
|
|
setattr(obj, name, old)
|
|
|
|
|
|
def test_dual_orchestration_order_auto_before_golden():
|
|
"""编排序 = gen-only → smoke → ensure_play_spec → play(auto) → inject_golden → play(golden);
|
|
auto_gates 在金标注入之前捕获(两次 play 返不同 verdict 可区分)。"""
|
|
import cheap_studio
|
|
import cheap_run
|
|
import compare_node
|
|
calls = []
|
|
saved = []
|
|
with tempfile.TemporaryDirectory() as t:
|
|
td = Path(t)
|
|
|
|
async def fake_gen(gid, brief, **kw):
|
|
calls.append("gen")
|
|
|
|
plays = iter([
|
|
{"verdict": {"guards": {"E_live": {"pass": True}, "H_progress": {"pass": True}}}}, # auto
|
|
{"verdict": {"guards": {"E_live": {"pass": False}, "H_progress": {"pass": True}}}}, # golden
|
|
])
|
|
_patch(cheap_studio, "run_studio", fake_gen, saved)
|
|
_patch(cheap_run, "smoke", lambda gid, **kw: (calls.append("smoke") or {"ok": True, "state": {"targets": []}}), saved)
|
|
_patch(cheap_run, "ensure_play_spec", lambda gid, state: (calls.append("ensure") or {"wrote": True}), saved)
|
|
_patch(cheap_run, "play", lambda gid, **kw: (calls.append("play") or next(plays)), saved)
|
|
_patch(cheap_run, "wg1_game_dir", lambda gid: td / gid, saved) # staged 无 spec
|
|
_patch(cheap_run, "game_dir", lambda gid: td / gid, saved)
|
|
_patch(compare_node, "inject_golden", lambda gid, g: (calls.append("inject") or (td / "g")), saved)
|
|
_patch(compare_node, "golden_spec_path", lambda genre: td / "golden.json", saved)
|
|
_patch(compare_node, "src_shape", lambda d: {"realSrcMultifile": True}, saved)
|
|
try:
|
|
r = A.run_pair_dual_sync({"key": "click-score", "golden": "x.json"}, "brief", "avg-click-score-0")
|
|
finally:
|
|
_restore(saved)
|
|
|
|
assert calls == ["gen", "smoke", "ensure", "play", "inject", "play"]
|
|
# auto_gates 来自第一次 play(E_live=True),golden_gates 来自第二次(E_live=False)。
|
|
assert r["autoGates"]["E_live"] is True and r["goldenGates"]["E_live"] is False
|
|
assert r["gid"] == "avg-click-score-0" and r["realSrc"] is True
|
|
|
|
|
|
def test_dual_raises_when_staged_spec_exists():
|
|
"""staged 残留 play-spec.json → ensure 前置断言触发、不静默用旧 spec 驱动 auto。"""
|
|
import cheap_studio
|
|
import cheap_run
|
|
saved = []
|
|
with tempfile.TemporaryDirectory() as t:
|
|
td = Path(t)
|
|
gd = td / "avg-click-score-0"
|
|
gd.mkdir(parents=True)
|
|
(gd / "play-spec.json").write_text("{}", encoding="utf-8") # 残留旧 spec
|
|
|
|
async def fake_gen(gid, brief, **kw):
|
|
pass
|
|
|
|
_patch(cheap_studio, "run_studio", fake_gen, saved)
|
|
_patch(cheap_run, "smoke", lambda gid, **kw: {"ok": True, "state": {}}, saved)
|
|
_patch(cheap_run, "wg1_game_dir", lambda gid: gd, saved)
|
|
raised = False
|
|
try:
|
|
A.run_pair_dual_sync({"key": "click-score", "golden": "x.json"}, "brief", "avg-click-score-0")
|
|
except RuntimeError:
|
|
raised = True
|
|
finally:
|
|
_restore(saved)
|
|
assert raised is True
|
|
|
|
|
|
def test_clamp_conc_upper_cap():
|
|
"""并发上限夹取 ≤15(复用 bake_off 口径)。"""
|
|
assert A._clamp_conc(100) == 15 and A._clamp_conc(0) == 1 and A._clamp_conc(3) == 3
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
|
_failed = 0
|
|
for _fn in _fns:
|
|
try:
|
|
_fn()
|
|
print(f" PASS {_fn.__name__}")
|
|
except Exception as e: # noqa: BLE001
|
|
_failed += 1
|
|
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
|
|
print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
|
|
sys.exit(1 if _failed else 0)
|