209 lines
11 KiB
Python
209 lines
11 KiB
Python
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("""<testsuite>
|
|
<testcase name='mars_stream_save'><system-out>test_save: ok (2 save(s), 0 failures)</system-out></testcase>
|
|
<testcase name='mars_stream_domains'><system-out>domains: 2 save(s)</system-out></testcase>
|
|
<testcase name='app_turn'><system-out>app: 2 save(s)</system-out></testcase>
|
|
<testcase name='app_turn_record'><system-out>record: 2 save(s)</system-out></testcase>
|
|
<testcase name='app_turn_record'><system-out>record: 2 save(s)</system-out></testcase>
|
|
</testsuite>""")
|
|
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()
|