import importlib.util import json import subprocess import tempfile import unittest import sys from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / "verify/tooling")) from fixtures import passed_gate def load(name): spec = importlib.util.spec_from_file_location(name, ROOT / "tools" / (name + ".py")) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module gate = load("gate") evidence = load("evidence") reporter = load("standalone_report") class ToolingTests(unittest.TestCase): def passed_gate(self, binary, digest): return passed_gate(binary, digest) def test_inventory_is_explicit_and_unique(self): data = json.loads((ROOT / "verify/tooling/host-tests.json").read_text()) self.assertEqual(59, len(data["tests"])) self.assertEqual(len(data["tests"]), len(set(data["tests"]))) def test_build_named_source_is_not_excluded(self): with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "build_turn.cpp").write_text("source") result = subprocess.CompletedProcess([], 0, b"build_turn.cpp\0") with mock.patch.object(gate.subprocess, "run", return_value=result): self.assertEqual([Path("build_turn.cpp")], gate.files_for_git_tree(root)) def test_snapshot_rejects_pre_copy_mutation(self): with tempfile.TemporaryDirectory() as td: root, destination = Path(td) / "source", Path(td) / "copy" root.mkdir(); (root / "a.cpp").write_text("before") manifest = {"files": [gate.file_row(root, Path("a.cpp"))]} (root / "a.cpp").write_text("after") with self.assertRaisesRegex(ValueError, "changed before snapshot"): gate.copy_snapshot(root, destination, manifest) def test_source_symlink_is_rejected(self): with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "target").write_text("x") (root / "link").symlink_to(root / "target") with self.assertRaisesRegex(ValueError, "symlinks"): gate.file_row(root, Path("link")) def test_junit_requires_complete_unique_identity_and_positive_corpus_counts(self): with tempfile.TemporaryDirectory() as td: path = Path(td) / "ctest.xml" path.write_text(""" test_save: ok (2 save(s), 0 failures) domains: 2 save(s) app: 2 save(s) record: 2 save(s) record: 2 save(s) """) rows = gate.junit_statuses(path) self.assertEqual([2], gate.corpus_counts(rows)["mars_stream_save"]) names = [row["name"] for row in rows] self.assertNotEqual(len(names), len(set(names))) self.assertEqual([], [row for row in rows if row["systemOut"] is None]) def test_run_exception_is_a_recorded_failure(self): logs = [] with mock.patch.object(gate.subprocess, "run", side_effect=OSError("missing")): self.assertIsNone(gate.run(["missing"], ROOT, logs)) self.assertIn("missing", logs[0]["stderr"]) def test_passed_gate_schema_must_be_complete(self): with tempfile.TemporaryDirectory() as td: path = Path(td) / "manifest.json" path.write_text('{"schema":"sots-gate/1","status":"passed"}') with self.assertRaises(ValueError): evidence.load_passed_gate(path) def test_failed_gate_is_not_reporter_provenance(self): with tempfile.TemporaryDirectory() as td: path = Path(td) / "manifest.json" path.write_text('{"schema":"sots-gate/1","status":"failed"}') with self.assertRaises(ValueError): evidence.load_passed_gate(path) def test_reporter_rejects_output_and_roundtrip_override(self): with self.assertRaisesRegex(ValueError, "override"): reporter.checked_engine_args(["--out", "elsewhere"]) with self.assertRaisesRegex(ValueError, "override"): reporter.checked_engine_args(["--roundtrip-only"]) def test_reporter_hashes_actual_engine_file_inputs(self): with tempfile.TemporaryDirectory() as td: commands = Path(td) / "turn.tcb"; commands.write_text("capture") args, inputs = reporter.checked_engine_args(["--turn-commands", str(commands), "--commit-rng"]) self.assertEqual(["--turn-commands", str(commands), "--commit-rng"], args) self.assertEqual("file", inputs[0]["value"]["kind"]) def test_reporter_rejects_binary_mismatch_before_pair_execution(self): with tempfile.TemporaryDirectory() as td: root = Path(td) binary, provenance, out = root / "binary", root / "gate.json", root / "out" source, oracle = root / "source.sav", root / "oracle.sav" binary.write_text("binary"); source.write_text("source"); oracle.write_text("oracle") provenance.write_text(json.dumps(self.passed_gate(binary, "0" * 64))) with self.assertRaises(SystemExit): reporter.main(["--binary", str(binary), "--provenance", str(provenance), "--out", str(out), "--pair", str(source), str(oracle)]) def test_reporter_match_requirement_rejects_wrong_reference_and_pair_failure(self): class Checksum: def checksum_save(self, path): return type("Result", (), {"coverage": {"ok": True}, "digest": Path(path).read_text(), "root": Path(path).read_text()})() def diff(self, left, right, **kwargs): return [(left, right)] with tempfile.TemporaryDirectory() as td: root = Path(td) binary, provenance = root / "binary", root / "gate.json" source, oracle = root / "source.sav", root / "oracle.sav" binary.write_text("binary"); source.write_text("source"); oracle.write_text("oracle") provenance.write_text(json.dumps(self.passed_gate(binary, reporter.digest(binary)))) def wrong_reference(command, **_): Path(command[command.index("--out") + 1]).write_text("wrong") return subprocess.CompletedProcess(command, 0, "ran", "") with mock.patch.object(reporter, "checksum_module", return_value=Checksum()), \ mock.patch.object(reporter.subprocess, "run", side_effect=wrong_reference): self.assertEqual(1, reporter.main(["--binary", str(binary), "--provenance", str(provenance), "--out", str(root / "wrong"), "--pair", str(source), str(oracle), "--require-match"])) with mock.patch.object(reporter, "checksum_module", return_value=Checksum()), \ mock.patch.object(reporter.subprocess, "run", return_value=subprocess.CompletedProcess([], 1, "", "failed")): self.assertEqual(1, reporter.main(["--binary", str(binary), "--provenance", str(provenance), "--out", str(root / "failed"), "--pair", str(source), str(oracle)])) def test_reporter_requires_at_least_one_pair(self): with self.assertRaises(SystemExit): reporter.main(["--binary", "binary", "--provenance", "gate", "--out", "new"]) def test_stale_report_output_is_rejected_before_execution(self): with tempfile.TemporaryDirectory() as td: out = Path(td) / "out"; out.mkdir() with self.assertRaises(SystemExit): reporter.main(["--binary", str(ROOT / "no-binary"), "--provenance", str(ROOT / "no-gate"), "--out", str(out), "--pair", "a", "b"]) def test_jobs_must_be_positive(self): with self.assertRaises(SystemExit): gate.main(["--engine", str(ROOT), "--corpus", str(ROOT), "--out", "/tmp/nope", "--jobs", "0"]) def test_manifest_cannot_choose_its_own_required_checks(self): value = self.passed_gate("binary", "a" * 64) value["requiredChecks"] = ["binaryExists"] with self.assertRaisesRegex(ValueError, "omitted"): evidence.validate_gate(value) def test_empty_or_duplicate_corpus_execution_is_rejected(self): for counts in ([], [0], [1, 1]): value = self.passed_gate("binary", "a" * 64) value["tests"]["corpusSummaryCounts"]["app_turn"] = counts with self.assertRaisesRegex(ValueError, "positive corpus"): evidence.validate_gate(value) def test_corpus_count_uses_final_summary_not_incidental_mentions(self): rows = [{"name": "mars_stream_domains", "systemOut": "== value domains over 43 save(s)\ntest_domains: ok (43 save(s), 0 failure(s))\n"}] self.assertEqual([43], gate.corpus_counts(rows)["mars_stream_domains"]) rows[0]["systemOut"] += "test_domains: ok (43 save(s), 0 failure(s))\n" self.assertEqual([43, 43], gate.corpus_counts(rows)["mars_stream_domains"]) def test_equal_pair_is_measurement_and_output_is_retained(self): class Checksum: def checksum_save(self, path): return type("Result", (), {"coverage": {"ok": True}, "digest": "same", "root": "same"})() with tempfile.TemporaryDirectory() as td: root = Path(td) binary, provenance, source = root / "binary", root / "gate.json", root / "input.sav" binary.write_text("executable"); source.write_text("same raw stream") provenance.write_text(json.dumps(self.passed_gate(binary, reporter.digest(binary)))) def copy_input(command, **kwargs): Path(command[command.index("--out") + 1]).write_bytes(source.read_bytes()) return subprocess.CompletedProcess(command, 0, "simulated", "") with mock.patch.object(reporter, "checksum_module", return_value=Checksum()), \ mock.patch.object(reporter.subprocess, "run", side_effect=copy_input): result = reporter.main(["--binary", str(binary), "--provenance", str(provenance), "--out", str(root / "result"), "--pair", str(source), str(source), "--require-match"]) self.assertEqual(result, 0) manifest = json.loads((root / "result/manifest.json").read_text()) self.assertEqual(manifest["status"], "measured") self.assertTrue(manifest["allPairsMatch"]) self.assertTrue(Path(manifest["pairs"][0]["output"]).is_file()) if __name__ == "__main__": unittest.main()