import copy import importlib.util import json import os from pathlib import Path import signal import subprocess import sys import tempfile import time import unittest from unittest import mock SCRIPT = Path(__file__).with_name("verify.py") SPEC = importlib.util.spec_from_file_location("verify", SCRIPT) verify = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(verify) class VerificationTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.directory = Path(self.temporary.name) def test_contract_cannot_remove_whole_proof_or_interpreter_gates(self): contract, _ = verify.load_contract() loader = verify.tomllib.load for section, field, value in ( (None, "properties", []), ("miri", "packages", []), ("miri", "packages", ["kaptan-log", "kaptan-log"]), ("miri", "seeds", []), ("miri", "seeds", [0, 0]), ("miri", "seeds", [-1]), ("miri", "seeds", [True]), ): damaged = copy.deepcopy(contract) (damaged[section] if section else damaged)[field] = value def load(source): if Path(source.name).name == "verification.toml": return damaged return loader(source) with ( self.subTest(field=field, value=value), mock.patch.object(verify.tomllib, "load", side_effect=load), self.assertRaises(ValueError), ): verify.load_contract() def proof(self): prop = { "id": "example", "harnesses": ["proof"], "covers": [1], "package": "example", "source": "proof.rs", } report = { "metadata": { "version": "1.0", "kani_version": "0.68.0", "target": "x86_64-unknown-linux-gnu", }, "tools": {"rustc": "pinned rustc"}, "harness_metadata": [ { "pretty_name": "proof", "crate_name": "example", "source": {"file": "proof.rs"}, } ], "verification_results": { "summary": { "total_harnesses": 1, "executed": 1, "status": "completed", "successful": 1, "failed": 0, }, "results": [ { "harness_id": "proof", "status": "Success", "checks": [ { "category": "assertion", "status": "Success", "description": "assertion failed: invariant", "function": "proof", }, {"category": "cover", "status": "Satisfied"}, ], } ], }, } return prop, report def validate(self, prop, report, exit_code=0): path = self.directory / "proof.json" path.write_text(json.dumps(report)) return verify.validate_kani( path, prop, "0.68.0", "x86_64-unknown-linux-gnu", exit_code, "pinned rustc" ) def test_failed_missing_vacuous_and_unknown_proofs_cannot_pass(self): prop, report = self.proof() self.assertTrue(self.validate(prop, report)["raw_verifier_passed"]) for status in ( "Failure", "Undetermined", "SolverError", "Unreachable", "new-status", ): damaged = copy.deepcopy(report) damaged["verification_results"]["results"][0]["checks"][0]["status"] = ( status ) with self.subTest(status=status), self.assertRaises(ValueError): self.validate(prop, damaged) for status in ("Unreachable", "Unsatisfiable", "Undetermined"): damaged = copy.deepcopy(report) damaged["verification_results"]["results"][0]["checks"][1]["status"] = ( status ) with self.subTest(cover=status), self.assertRaises(ValueError): self.validate(prop, damaged) damaged = copy.deepcopy(report) damaged["verification_results"]["results"][0]["checks"].pop() with self.assertRaises(ValueError): self.validate(prop, damaged) damaged["verification_results"]["results"] = [] with self.assertRaises(ValueError): self.validate(prop, damaged) with self.assertRaises(ValueError): self.validate(prop, report, exit_code=1) damaged = copy.deepcopy(report) damaged["harness_metadata"][0]["source"]["file"] = "another.rs" with self.assertRaises(ValueError): self.validate(prop, damaged) damaged = copy.deepcopy(report) damaged["tools"]["rustc"] = "different compiler" with self.assertRaises(ValueError): self.validate(prop, damaged) def test_pid_exception_preserves_raw_failure_and_rejects_other_failures(self): prop, report = self.proof() prop.update( id="control.pid-state", harnesses=["kani_proofs::pid_update_preserves_bounds_or_state"], intermediate_nan_line=196, ) result = report["verification_results"]["results"][0] result.update(harness_id=prop["harnesses"][0], status="Failure") report["harness_metadata"][0]["pretty_name"] = prop["harnesses"][0] report["verification_results"]["summary"].update(successful=0, failed=1) diagnostic = { "category": "NaN", "status": "Failure", "description": "NaN on addition", "function": "Pid::update_with_valid_time", "location": { "file": "libs/kaptan-control/src/lib.rs", "line": "196", "column": "13", }, } result["checks"].append(diagnostic) accepted = self.validate(prop, report, exit_code=1) self.assertFalse(accepted["raw_verifier_passed"]) self.assertTrue(accepted["contract_checks_passed"]) for field, value in ( ("category", "overflow"), ("description", "NaN on division"), ("function", "different"), ): damaged = copy.deepcopy(report) damaged["verification_results"]["results"][0]["checks"][-1][field] = value with self.subTest(field=field), self.assertRaises(ValueError): self.validate(prop, damaged, exit_code=1) result["checks"].append(copy.deepcopy(diagnostic)) with self.assertRaises(ValueError): self.validate(prop, report, exit_code=1) def test_miri_cannot_pass_with_empty_ignored_or_filtered_suites(self): path = self.directory / "miri.log" template = "test result: ok. {} passed; 0 failed; {} ignored; 0 measured; {} filtered out" for passed, ignored, filtered in ((0, 0, 0), (1, 1, 0), (1, 0, 1)): path.write_text(template.format(passed, ignored, filtered)) with self.assertRaises(ValueError): verify.validate_miri(path, 1) path.write_text(template.format(4, 0, 0)) self.assertEqual(verify.validate_miri(path, 1), 4) with self.assertRaises(ValueError): verify.validate_miri(path, 2) def test_snapshot_includes_untracked_bytes_modes_and_deletions(self): path = self.directory / "untracked.rs" path.write_text("original") with mock.patch.object( verify.subprocess, "check_output", return_value=b"untracked.rs\0deleted.rs\0", ): before = verify.snapshot(self.directory) path.write_text("edited") self.assertNotEqual(before, verify.snapshot(self.directory)) path.write_text("original") path.chmod(0o755) self.assertNotEqual(before, verify.snapshot(self.directory)) path.unlink() self.assertTrue( all( entry["kind"] == "absent" for entry in verify.snapshot(self.directory)["entries"] ) ) def test_compiler_and_target_environment_overrides_are_rejected(self): for name in ( "RUSTFLAGS", "MIRIFLAGS", "CARGO_BUILD_RUSTFLAGS", "CARGO_BUILD_RUSTC_WRAPPER", "CARGO_ENCODED_RUSTDOCFLAGS", "CARGO_PROFILE_RELEASE_OVERFLOW_CHECKS", "CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS", ): with self.subTest(name=name), self.assertRaises(ValueError): verify.reject_overrides({name: "override"}) verify.reject_overrides({"PATH": "/bin", "CARGO_TARGET_DIR": "/tmp/build"}) def test_make_controls_cannot_skip_gate_recipes_or_override_commands(self): (self.directory / "Makefile").write_text( "COMMAND = touch executed\n.PHONY: gate\ngate:\n\t$(COMMAND)\n" ) injected = self.directory / "injected.mk" injected.write_text("override COMMAND = true\n") for controls in ( {"MAKEFLAGS": "n"}, {"MAKEFLAGS": "t"}, {"MFLAGS": "-n", "GNUMAKEFLAGS": "-n"}, {"MAKEFLAGS": "-- COMMAND=true", "MAKEOVERRIDES": "COMMAND=true"}, {"MAKEFILES": str(injected)}, ): with self.subTest(controls=controls): marker = self.directory / "executed" marker.unlink(missing_ok=True) result = verify.run( ["make", "-C", str(self.directory), "gate"], self.directory / "make.log", verify.make_environment(os.environ | controls), 5, ) self.assertTrue(result["passed"]) self.assertTrue(marker.exists()) def test_optimized_final_link_rejects_invalid_outcomes(self): fixture = ( SCRIPT.parents[1] / "libs/kaptan-err-reportable-macros/tests/final_link.py" ) harness = """ import importlib.util,subprocess,sys spec=importlib.util.spec_from_file_location('final_link',sys.argv[1]) fixture=importlib.util.module_from_spec(spec); spec.loader.exec_module(fixture) failure=sys.argv[3] def run(arguments, cwd): linking=arguments[0]=='c++' or arguments[-1]=='error-id-app' duplicate=any('used-duplicate' in arg for arg in arguments) code=int(linking and duplicate and failure=='diagnostic') return subprocess.CompletedProcess(arguments,code, 'host: test-host\\n' if arguments[0]=='rustc' else '', 'unrelated link failure' if code else '') fixture.run=run sys.argv=[sys.argv[1],'--work-dir',sys.argv[2]] fixture.main() """ for case, diagnostic in ( ("success", "duplicate unexpectedly linked"), ("diagnostic", "unrelated link failure"), ): with self.subTest(case=case): result = subprocess.run( [ sys.executable, "-O", "-c", harness, str(fixture), str(self.directory / case), case, ], capture_output=True, text=True, timeout=10, ) self.assertNotEqual(result.returncode, 0) self.assertIn(diagnostic, result.stderr) def test_source_intake_drift_and_finalization_failures_leave_failed_receipts(self): baseline = {"sha256": "baseline", "entries": []} changed = {"sha256": "changed", "entries": []} for stage, snapshots in ( ("intake", [ValueError("unavailable source")]), ("archive", [baseline, baseline]), ("drift", [baseline, baseline, changed, changed]), ("finalization", [baseline, baseline, baseline, OSError("final snapshot")]), ): output = self.directory / stage result = { "completed": True, "passed": True, "exit_code": 0, "log": "unused", } with ( self.subTest(stage=stage), mock.patch.object( sys, "argv", [str(SCRIPT), "host", "--output", str(output)] ), mock.patch.object(verify.signal, "signal"), mock.patch.object(verify, "snapshot", side_effect=snapshots), mock.patch.object(verify, "capture", return_value="identity"), mock.patch.object(verify, "run", return_value=result), mock.patch.object( verify.tarfile, "open", wraps=verify.tarfile.open, side_effect=OSError("archive") if stage == "archive" else None, ), ): self.assertEqual(verify.main(), 1) receipt = json.loads((output / "receipt.json").read_text()) self.assertFalse(receipt["passed"]) if stage != "archive": self.assertFalse(receipt["source_stable"]) def test_launch_failure_and_timeout_are_recorded(self): failed = verify.run( ["/no-such-verification-program"], self.directory / "missing.log", os.environ, 1, ) self.assertFalse(failed["completed"]) timed = verify.run( [sys.executable, "-c", "import time; time.sleep(30)"], self.directory / "timeout.log", os.environ, 0.05, ) self.assertFalse(timed["passed"]) self.assertFalse(timed["completed"]) def test_repeated_termination_kills_the_owned_process_group(self): pid_file = self.directory / "child.pid" receipt = self.directory / "result.json" child = ( "import os,time,pathlib; pathlib.Path(%r).write_text(str(os.getpid())); time.sleep(30)" % str(pid_file) ) harness = """ import importlib.util,json,os,pathlib,signal,sys spec=importlib.util.spec_from_file_location('verify',sys.argv[1]) v=importlib.util.module_from_spec(spec); spec.loader.exec_module(v) for s in (signal.SIGINT,signal.SIGTERM): signal.signal(s,v.interrupt) r=v.run([sys.executable,'-c',sys.argv[2]],pathlib.Path(sys.argv[3]),os.environ,30) pathlib.Path(sys.argv[4]).write_text(json.dumps(r)) """ parent = subprocess.Popen( [ sys.executable, "-c", harness, str(SCRIPT), child, str(self.directory / "signal.log"), str(receipt), ] ) def cleanup(): if pid_file.exists(): try: os.killpg(int(pid_file.read_text()), signal.SIGKILL) except ProcessLookupError: pass if parent.poll() is None: parent.kill() parent.wait() self.addCleanup(cleanup) deadline = time.monotonic() + 5 while not pid_file.exists() and time.monotonic() < deadline: time.sleep(0.01) self.assertTrue(pid_file.exists()) child_pid = int(pid_file.read_text()) os.kill(parent.pid, signal.SIGTERM) os.kill(parent.pid, signal.SIGTERM) self.assertEqual(parent.wait(timeout=5), 0) self.assertFalse(json.loads(receipt.read_text())["completed"]) with self.assertRaises(ProcessLookupError): os.kill(child_pid, 0) def test_cancellation_during_spawn_successful_wait_or_cleanup_cannot_pass(self): for stage in ("spawn", "wait", "cleanup"): process = mock.Mock(pid=999999) def wait(**_kwargs): if stage == "wait": os.kill(os.getpid(), signal.SIGTERM) return 0 def spawn(*_args, **_kwargs): if stage == "spawn": os.kill(os.getpid(), signal.SIGTERM) return process def kill(*_args): if stage == "cleanup": os.kill(os.getpid(), signal.SIGTERM) process.wait.side_effect = wait with ( self.subTest(stage=stage), mock.patch.object(verify.subprocess, "Popen", side_effect=spawn), mock.patch.object(verify.os, "killpg", side_effect=kill), ): result = verify.run( ["fake-child"], self.directory / f"{stage}.log", os.environ, 1 ) self.assertFalse(result["passed"]) self.assertFalse(result["completed"]) process.wait.assert_called() if __name__ == "__main__": unittest.main()