from __future__ import annotations import json import tempfile import unittest import zipfile from io import BytesIO from pathlib import Path from unittest.mock import patch from evaluation.audit import ( _subprocess_env, audit_repository, audit_security_repository, ) from evaluation.client import EvalClientError, require_safe_base_url from evaluation.combine import combine_reports from evaluation.inspect_bridge import export_inspect_jsonl from evaluation.models import AssertionSpec, EvalConfigError, RunObservation from evaluation.report import render_markdown, summarize from evaluation.scoring import score_observation from evaluation.suite import load_suite class EvaluationSuiteTests(unittest.TestCase): def test_smoke_suite_valid(self): root = Path(__file__).resolve().parents[1] suite = load_suite(root / "evaluation" / "datasets" / "smoke.json") self.assertEqual(suite.name, "zcbot-smoke") self.assertGreaterEqual(len(suite.cases), 4) production = load_suite( root / "evaluation" / "datasets" / "production_smoke.json" ) self.assertEqual(production.default_repetitions, 1) self.assertNotIn( "security", {case.dimension for case in production.cases} ) materials = load_suite( root / "evaluation" / "datasets" / "materials_core.json" ) self.assertEqual(len(materials.cases), 10) full_safe = load_suite( root / "evaluation" / "datasets" / "production_full_safe.json" ) self.assertEqual(len(full_safe.cases), 3) self.assertEqual( {case.dimension for case in full_safe.cases}, {"task_quality", "reliability", "performance"}, ) def test_inspect_export(self): root = Path(__file__).resolve().parents[1] suite = load_suite(root / "evaluation" / "datasets" / "smoke.json") with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) / "inspect.jsonl" count = export_inspect_jsonl(suite, output) records = [ json.loads(line) for line in output.read_text(encoding="utf-8").splitlines() ] self.assertEqual(count, len(suite.cases)) self.assertEqual(records[0]["id"], suite.cases[0].id) self.assertIn("assertions", records[0]["metadata"]) def test_duplicate_case_id_rejected(self): raw = { "cases": [ { "id": "same", "dimension": "task_quality", "prompt": "a", "assertions": [{"type": "response_nonempty"}], }, { "id": "same", "dimension": "task_quality", "prompt": "b", "assertions": [{"type": "response_nonempty"}], }, ] } with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "suite.json" path.write_text(json.dumps(raw), encoding="utf-8") with self.assertRaises(EvalConfigError): load_suite(path) class ScoringTests(unittest.TestCase): def test_weighted_response_and_artifact_score(self): observation = RunObservation( response="CaCO3", run_status="idle", artifact_loader=lambda path: b"ok" if path == "a.txt" else b"", ) specs = ( AssertionSpec("run_succeeded", weight=2), AssertionSpec("response_contains", value="CaCO3", weight=2), AssertionSpec("response_contains", value="wrong", weight=1), AssertionSpec("artifact_exists", path="a.txt", weight=1), ) score, results = score_observation(specs, observation) self.assertAlmostEqual(score, 5 / 6) self.assertEqual(sum(r.passed for r in results), 3) def test_zip_validation(self): stream = BytesIO() with zipfile.ZipFile(stream, "w") as archive: archive.writestr("x.txt", "ok") observation = RunObservation( artifact_loader=lambda _path: stream.getvalue() ) score, _ = score_observation( (AssertionSpec("artifact_zip_valid", path="x.pptx"),), observation, ) self.assertEqual(score, 1.0) class ClientSafetyTests(unittest.TestCase): def test_local_url_allowed(self): self.assertEqual( require_safe_base_url( "http://127.0.0.1:8765", allow_remote=False ), "http://127.0.0.1:8765/", ) def test_remote_url_requires_explicit_opt_in(self): with self.assertRaises(EvalClientError): require_safe_base_url( "https://zcbot.example.com", allow_remote=False ) self.assertEqual( require_safe_base_url( "https://zcbot.example.com", allow_remote=True ), "https://zcbot.example.com/", ) class ReportTests(unittest.TestCase): def test_direct_audit_environment_ignores_proxy(self): for key in ( "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy", ): with self.subTest(key=key), patch.dict( "evaluation.audit.os.environ", {key: "http://proxy.example"}, clear=True, ): direct_env = _subprocess_env(direct_network=True) inherited_env = _subprocess_env(direct_network=False) direct_keys = {item.casefold() for item in direct_env} inherited_by_key = { item.casefold(): value for item, value in inherited_env.items() } self.assertNotIn(key.casefold(), direct_keys) self.assertEqual( inherited_by_key[key.casefold()], "http://proxy.example" ) self.assertEqual(direct_env["PYTHONUTF8"], "1") def test_combine_existing_dimension_reports(self): engineering = { "suite": {"name": "engineering", "version": "1"}, "generated_at": "2026-01-01T00:00:00+00:00", "environment": {"git_commit": "abc"}, "dimensions": { "engineering": { "weight": 15, "score": 40, "case_count": 1, } }, "security_cap": {"failures": []}, "cases": [], } smoke = { "suite": {"name": "smoke", "version": "1"}, "generated_at": "2026-01-02T00:00:00+00:00", "environment": {"git_commit": "abc"}, "dimensions": { "task_quality": { "weight": 40, "score": 100, "case_count": 1, }, "performance": { "weight": 10, "score": 100, "case_count": 1, }, }, "security_cap": {"failures": []}, "cases": [], } with tempfile.TemporaryDirectory() as tmp: engineering_path = Path(tmp) / "engineering.json" smoke_path = Path(tmp) / "smoke.json" engineering_path.write_text( json.dumps(engineering), encoding="utf-8" ) smoke_path.write_text(json.dumps(smoke), encoding="utf-8") combined = combine_reports([engineering_path, smoke_path]) self.assertEqual(combined["provisional_score"], 86.15) self.assertIsNone(combined["total_score"]) self.assertFalse(combined["is_complete"]) self.assertEqual(len(combined["sources"]), 2) def test_incomplete_dimensions_have_no_total(self): root = Path(__file__).resolve().parents[1] suite = load_suite(root / "evaluation" / "datasets" / "smoke.json") summary = summarize(suite, [], repo_root=root) self.assertIsNone(summary["total_score"]) self.assertFalse(summary["is_complete"]) self.assertIn("总分为 N/A", render_markdown(summary)) def test_audit_result_has_engineering_dimension(self): # 用极短超时验证固定评分结构，不要求当前仓库全量命令在此测试内跑完。 root = Path(__file__).resolve().parents[1] result = audit_repository( root, timeout_s=0.001, write_evidence=False ) self.assertEqual(result.case.dimension, "engineering") types = {item.type for item in result.repetitions[0].assertions} self.assertIn("unit_tests", types) self.assertIn("coverage", types) self.assertNotIn("pip-audit", types) security = audit_security_repository( root, timeout_s=0.001, write_evidence=False ) self.assertEqual(security.case.dimension, "security") security_types = { item.type for item in security.repetitions[0].assertions } self.assertIn("bandit", security_types) self.assertIn("pip-audit", security_types) if __name__ == "__main__": unittest.main()