"""评估闭环 v7 单测：金任务校验、确定性评分、两轮对比、CLI。 运行（在仓库根目录下）： python3 -m unittest discover -s cloud/tests -p "test_eval_loop.py" -v 说明：纯函数 + CLI 子进程，不跑模型、不联网、不写库。 关键断言：未测定检查不计入分母、不当通过或失败； 配对差值只算两轮都有的金任务；配对样本 < 20 时 ready_for_decision 为 false；重复 task_id 直接报错； CLI 只读文件，非法输入退出 2。 """ import json import os import subprocess import sys import tempfile import unittest from pathlib import Path try: from cloud.eval_loop import compare_runs, score_task from cloud.eval_loop import validate_golden_tasks except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.eval_loop import compare_runs, score_task from cloud.eval_loop import validate_golden_tasks def _goldens(): return validate_golden_tasks([ {"id": "g1", "prompt": "算 17*23", "tier": "default", "checks": [{"type": "status_completed"}, {"type": "contains", "value": "391"}, {"type": "max_tokens", "value": 5000}]}, {"id": "g2", "prompt": "写函数", "tier": "default", "checks": [{"type": "status_completed"}, {"type": "not_contains", "value": "TODO"}]}, ]) def _record(task_id, golden_id, status="completed", result_text="391", tokens=100, micros=1000): return {"task_id": task_id, "golden_id": golden_id, "model": "m", "status": status, "result_text": result_text, "total_tokens": tokens, "usd_micros": micros, "unpriced_tokens": 0} class ValidateTestCase(unittest.TestCase): def test_rejects_bad_golden(self): for bad in ( [], [{"id": "g1", "prompt": "x", "checks": []}], [{"id": "g1", "prompt": "x", "checks": [{"type": "judge", "value": 1}]}], [{"id": "g1", "prompt": "x", "checks": [{"type": "contains", "value": "a"}]}, {"id": "g1", "prompt": "y", "checks": [{"type": "contains", "value": "b"}]}], [{"id": "g1", "prompt": " ", "checks": [ {"type": "status_completed"}]}], ): with self.assertRaises(ValueError, msg=str(bad)): validate_golden_tasks(bad) class ScoreTestCase(unittest.TestCase): def test_all_pass_and_partial(self): goldens = _goldens() scored = score_task(goldens["g1"], _record("t1", "g1")) self.assertTrue(scored["all_pass"]) self.assertEqual(scored["score"], 1.0) failed = score_task( goldens["g1"], _record("t2", "g1", result_text="392")) self.assertFalse(failed["all_pass"]) self.assertAlmostEqual(failed["score"], 2 / 3) def test_unknown_excluded_from_denominator(self): goldens = _goldens() scored = score_task( goldens["g1"], _record("t1", "g1", tokens=None, micros=None)) # status + contains 通过，max_tokens 未测定：分母 2 不是 3。 self.assertEqual(scored["score"], 1.0) self.assertEqual(scored["unknown"], ["max_tokens"]) # all_pass 的口径是“全部已测定检查通过”，含 unknown # 的任务仍可能为真，故须同时看 fully_measured。 self.assertTrue(scored["all_pass"]) self.assertFalse(scored["fully_measured"]) def test_empty_result_is_fail_not_unknown(self): goldens = _goldens() scored = score_task( goldens["g1"], _record("t1", "g1", result_text=None)) self.assertIn("contains", scored["failed"]) class CompareTestCase(unittest.TestCase): def test_paired_delta_and_small_sample_gate(self): goldens = _goldens() report = compare_runs( goldens, [_record("a1", "g1"), _record("a2", "g2", result_text="ok")], [_record("b1", "g1", result_text="392"), _record("b2", "g2", result_text="TODO")], label_a="A", label_b="B") self.assertEqual(report["paired_n"], 2) self.assertEqual(report["paired_pass_rate_a"], 1.0) self.assertEqual(report["paired_pass_rate_b"], 0.0) self.assertEqual(report["paired_pass_rate_delta"], -1.0) self.assertFalse(report["ready_for_decision"]) self.assertTrue(any("not enough" in w for w in report["warnings"])) def test_unmatched_excluded_and_missing_listed(self): goldens = _goldens() report = compare_runs( goldens, [_record("a1", "g1"), _record("a9", "nope")], [_record("b1", "g1")]) self.assertEqual(report["run_a"]["n"], 1) self.assertEqual(report["run_a"]["unmatched_n"], 1) self.assertEqual(report["missing_in_a"], ["g2"]) self.assertEqual(report["missing_in_b"], ["g2"]) def test_duplicate_task_id_raises(self): goldens = _goldens() with self.assertRaises(ValueError): compare_runs( goldens, [_record("a1", "g1"), _record("a1", "g2")], [_record("b1", "g1")]) class CliTestCase(unittest.TestCase): def test_cli_report_and_bad_input(self): repo_root = str(Path(__file__).resolve().parents[2]) with tempfile.TemporaryDirectory() as tmpdir: golden_path = os.path.join(tmpdir, "golden.json") run_a_path = os.path.join(tmpdir, "a.json") run_b_path = os.path.join(tmpdir, "b.json") with open(golden_path, "w", encoding="utf-8") as fh: json.dump([{"id": "g1", "prompt": "算 17*23", "checks": [{"type": "contains", "value": "391"}]}], fh) with open(run_a_path, "w", encoding="utf-8") as fh: json.dump([_record("a1", "g1")], fh) with open(run_b_path, "w", encoding="utf-8") as fh: json.dump([_record("b1", "g1", result_text="392")], fh) good = subprocess.run( [sys.executable, "-m", "cloud.eval_loop", "--golden", golden_path, "--run-a", run_a_path, "--run-b", run_b_path, "--label-a", "A", "--label-b", "B"], cwd=repo_root, capture_output=True, text=True, timeout=30) self.assertEqual(good.returncode, 0, good.stderr) report = json.loads(good.stdout) self.assertEqual(report["paired_n"], 1) self.assertEqual(report["paired_pass_rate_a"], 1.0) with open(golden_path, "w", encoding="utf-8") as fh: json.dump([{"id": "g1", "prompt": "x", "checks": []}], fh) bad = subprocess.run( [sys.executable, "-m", "cloud.eval_loop", "--golden", golden_path, "--run-a", run_a_path, "--run-b", run_b_path], cwd=repo_root, capture_output=True, text=True, timeout=30) self.assertEqual(bad.returncode, 2)