from __future__ import annotations import copy import importlib.machinery import importlib.util import json from pathlib import Path import re import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[2] LOADER = importlib.machinery.SourceFileLoader( "lazaret_verify", str(ROOT / "tools/lazaret_verify.py") ) SPEC = importlib.util.spec_from_loader(LOADER.name, LOADER) VERIFY = importlib.util.module_from_spec(SPEC) LOADER.exec_module(VERIFY) CORPUS_LOADER = importlib.machinery.SourceFileLoader( "ignored_workspace_verifier", str(ROOT / "tools/ignored_workspace_verifier.py") ) CORPUS_SPEC = importlib.util.spec_from_loader(CORPUS_LOADER.name, CORPUS_LOADER) CORPUS = importlib.util.module_from_spec(CORPUS_SPEC) CORPUS_LOADER.exec_module(CORPUS) class VerificationTests(unittest.TestCase): def test_client_implementation_dependency_boundary_is_enforced(self) -> None: metadata = VERIFY.workspace_metadata() client = next( package for package in metadata["packages"] if package["name"] == "lazaret-client" ) client["dependencies"].append( { "name": "lazaret-host-coordinator", "source": None, "path": str(ROOT / "crates" / "lazaret-host-coordinator"), "req": "=0.0.0", } ) with self.assertRaisesRegex(VERIFY.VerificationError, "Client depends"): VERIFY.audit_dependencies(metadata) def test_verification_temporaries_use_the_ignored_target_tree(self) -> None: original = VERIFY.VERIFICATION_TEMP_ROOT try: VERIFY.VERIFICATION_TEMP_ROOT = ROOT / "target" / "verification-test" with VERIFY.temporary_directory("example-") as directory: self.assertEqual(Path(directory).parent, VERIFY.VERIFICATION_TEMP_ROOT) finally: VERIFY.VERIFICATION_TEMP_ROOT = original def test_current_workspace_lints_pass(self) -> None: VERIFY.audit_workspace_lints(VERIFY.workspace_metadata()) def test_workspace_unsafe_policy_is_required(self) -> None: workspace = VERIFY.load_toml(ROOT / "Cargo.toml") workspace["workspace"]["lints"]["rust"]["unsafe_code"] = "deny" original = VERIFY.load_toml VERIFY.load_toml = lambda path: workspace if path == ROOT / "Cargo.toml" else original(path) try: with self.assertRaisesRegex(VERIFY.VerificationError, "forbid unsafe"): VERIFY.audit_workspace_lints(VERIFY.workspace_metadata()) finally: VERIFY.load_toml = original def test_clippy_process_policy_is_exact(self) -> None: config = VERIFY.load_toml(ROOT / ".clippy.toml") config["disallowed-methods"] = config["disallowed-methods"][:-1] original = VERIFY.load_toml VERIFY.load_toml = lambda path: config if path == ROOT / ".clippy.toml" else original(path) try: with self.assertRaisesRegex(VERIFY.VerificationError, "reviewed boundary"): VERIFY.audit_workspace_lints(VERIFY.workspace_metadata()) finally: VERIFY.load_toml = original def test_member_lint_inheritance_is_required(self) -> None: metadata = VERIFY.workspace_metadata() member = next( package for package in metadata["packages"] if package["id"] in set(metadata["workspace_members"]) ) original = VERIFY.load_toml def without_inheritance(path: Path) -> dict: data = original(path) if path == Path(member["manifest_path"]): data = copy.deepcopy(data) data.pop("lints") return data VERIFY.load_toml = without_inheritance try: with self.assertRaisesRegex(VERIFY.VerificationError, "inherit workspace lints"): VERIFY.audit_workspace_lints(metadata) finally: VERIFY.load_toml = original def test_lint_bypass_pattern_covers_restricted_lints(self) -> None: forbidden = ( "#[allow(disallowed_methods)]", "#![expect(clippy::disallowed_types)]", "#[allow(warnings)]", "#[allow(clippy::all)]", "#[allow(clippy::style)]", "#[expect(clippy::restriction)]", "#![allow(clippy :: all)]", "#[expect(clippy :: style)]", ) for source in forbidden: with self.subTest(source=source): self.assertIsNotNone(VERIFY.LINT_BYPASS.search(source)) self.assertIsNone(VERIFY.LINT_BYPASS.search("#[allow(dead_code)]")) def test_process_lint_allowances_are_exact_reviewed_functions(self) -> None: self.assertEqual( VERIFY.PROCESS_LINT_ALLOWANCES["crates/lazaret-linux/tests/capability-drop-helper.rs"], {"main"}, ) self.assertNotIn("spawn_unreviewed", set().union(*VERIFY.PROCESS_LINT_ALLOWANCES.values())) def test_process_lint_allowance_parser_rejects_false_exemptions(self) -> None: with self.assertRaisesRegex(VERIFY.VerificationError, "whole-file"): VERIFY.process_lint_allowances( "example.rs", "#![allow(clippy::disallowed_methods)]\nfn reviewed() {}" ) with self.assertRaisesRegex(VERIFY.VerificationError, "non-function"): VERIFY.process_lint_allowances( "example.rs", "#[allow(clippy::disallowed_methods)]\nstruct Boundary;\nfn hidden() {}", ) self.assertEqual( VERIFY.process_lint_allowances( "example.rs", "#[allow(clippy::disallowed_methods)]\nfn arbitrary() {}" ), {"arbitrary"}, ) def test_license_audit_accepts_permitted_choice_only(self) -> None: allowed = {"Apache-2.0", "MIT"} self.assertTrue( VERIFY.license_permitted( "Apache-2.0 WITH LLVM-exception OR Apache-2.0 OR MIT", allowed ) ) self.assertTrue(VERIFY.license_permitted("GPL-3.0-only OR MIT", allowed)) self.assertTrue(VERIFY.license_permitted("MIT/Apache-2.0", allowed)) self.assertTrue(VERIFY.license_permitted("Apache-2.0 / MIT", allowed)) self.assertFalse(VERIFY.license_permitted("GPL-3.0-only", allowed)) self.assertFalse(VERIFY.license_permitted("MIT AND GPL-3.0-only", allowed)) def test_only_the_exact_materialized_fuser_source_is_approved(self) -> None: package = { "name": "fuser", "version": "0.17.0", "manifest_path": str(VERIFY.MATERIALIZED_FUSER_MANIFEST), } self.assertTrue(VERIFY.is_materialized_fuser(package)) self.assertFalse(VERIFY.is_materialized_fuser({**package, "version": "0.18.0"})) self.assertFalse( VERIFY.is_materialized_fuser( {**package, "manifest_path": str(ROOT / "fuser/Cargo.toml")} ) ) def test_workspace_source_discovery_is_not_confined_to_crates(self) -> None: with tempfile.TemporaryDirectory(dir=ROOT) as directory: package_root = Path(directory) source = package_root / "outside.rs" source.write_text("fn main() {}") metadata = { "workspace_members": ["outside"], "packages": [ { "id": "outside", "manifest_path": str(package_root / "Cargo.toml"), } ], "target_directory": str(ROOT / "target"), } self.assertEqual(VERIFY.workspace_rust_sources(metadata), [source]) def test_workspace_source_discovery_rejects_overlapping_target_directory(self) -> None: with tempfile.TemporaryDirectory(dir=ROOT) as directory: package_root = Path(directory) metadata = { "workspace_members": ["outside"], "packages": [ { "id": "outside", "manifest_path": str(package_root / "Cargo.toml"), } ], "target_directory": str(ROOT), } with self.assertRaisesRegex(VERIFY.VerificationError, "must not contain"): VERIFY.workspace_rust_sources(metadata) metadata["target_directory"] = str(package_root) with self.assertRaisesRegex(VERIFY.VerificationError, "must not contain"): VERIFY.workspace_rust_sources(metadata) def test_rust_file_size_audit_rejects_unreviewed_oversized_source(self) -> None: with tempfile.TemporaryDirectory(dir=ROOT) as directory: package_root = Path(directory) source = package_root / "oversized.rs" source.write_text("fn item() {}\n" * (VERIFY.RUST_FILE_LINE_LIMIT + 1)) metadata = { "workspace_members": ["outside"], "packages": [ { "id": "outside", "manifest_path": str(package_root / "Cargo.toml"), } ], "target_directory": str(ROOT / "target"), } original = VERIFY.RUST_FILE_SIZE_EXCEPTIONS try: VERIFY.RUST_FILE_SIZE_EXCEPTIONS = {} with self.assertRaisesRegex(VERIFY.VerificationError, "oversized.rs"): VERIFY.audit_rust_file_sizes(metadata) finally: VERIFY.RUST_FILE_SIZE_EXCEPTIONS = original def test_rust_module_boundary_audit_rejects_lexical_include(self) -> None: with tempfile.TemporaryDirectory(dir=ROOT) as directory: package_root = Path(directory) (package_root / "lib.rs").write_text('include!("fragment.rs");\n') metadata = { "workspace_members": ["outside"], "packages": [ { "id": "outside", "manifest_path": str(package_root / "Cargo.toml"), } ], "target_directory": str(ROOT / "target"), } with self.assertRaisesRegex(VERIFY.VerificationError, "lexical include"): VERIFY.audit_rust_module_boundaries(metadata) def test_rust_module_boundary_audit_accepts_real_modules(self) -> None: with tempfile.TemporaryDirectory(dir=ROOT) as directory: package_root = Path(directory) (package_root / "lib.rs").write_text("mod fragment;\n") metadata = { "workspace_members": ["outside"], "packages": [ { "id": "outside", "manifest_path": str(package_root / "Cargo.toml"), } ], "target_directory": str(ROOT / "target"), } VERIFY.audit_rust_module_boundaries(metadata) def test_rust_file_size_exceptions_are_exact_and_current(self) -> None: metadata = VERIFY.workspace_metadata() original = VERIFY.RUST_FILE_SIZE_EXCEPTIONS cases = ( {"/absolute.rs": "one durable responsibility"}, {"missing.rs": "one durable responsibility"}, {"crates/lazaret-core/src/lib.rs": "one durable responsibility"}, {"crates/lazaret-core/src/configuration.rs": "vague"}, ) try: for exceptions in cases: with self.subTest(exceptions=exceptions): VERIFY.RUST_FILE_SIZE_EXCEPTIONS = exceptions with self.assertRaises(VERIFY.VerificationError): VERIFY.audit_rust_file_sizes(metadata) finally: VERIFY.RUST_FILE_SIZE_EXCEPTIONS = original def test_process_trace_requires_exactly_one_exec(self) -> None: VERIFY.validate_process_trace("example", ['execve("example", [], []) = 0']) with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 0"): VERIFY.validate_process_trace("example", []) with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 2"): VERIFY.validate_process_trace("example", ["execve(a)", "execve(b)"]) def test_wrapper_trace_requires_its_exec_in_place(self) -> None: VERIFY.validate_process_trace( "lazaret-launch-wrapper", ["execve(wrapper)", "execve(/bin/true)"], ) with self.assertRaisesRegex(VERIFY.VerificationError, "execve surface count 1"): VERIFY.validate_process_trace("lazaret-launch-wrapper", ["execve(wrapper)"]) def test_process_trace_rejects_each_forbidden_surface(self) -> None: forbidden = ( "clone(child_stack=NULL) = 1", "clone3(NULL, 0) = 1", "execveat(3, \"child\", [], [], 0) = 0", "fork() = 1", "vfork() = 1", "wait4(1, NULL, 0, NULL) = 1", "waitid(P_PIDFD, 3, {}, WEXITED, NULL) = 0", "rt_sigaction(SIGCHLD, {}, NULL, 8) = 0", ) for syscall in forbidden: with self.subTest(syscall=syscall): with self.assertRaisesRegex(VERIFY.VerificationError, "forbidden process syscall"): VERIFY.validate_process_trace("example", ["execve(example)", syscall]) def test_process_syscall_audit_requires_strace(self) -> None: with mock.patch.object(VERIFY.shutil, "which", return_value=None): with self.assertRaisesRegex(VERIFY.VerificationError, "strace must be installed"): VERIFY.audit_process_syscalls({}) def test_process_syscall_audit_rejects_failed_binary(self) -> None: for binary, returncode in VERIFY.PROCESS_TRACE_EXIT_CODES.items(): VERIFY.validate_trace_exit(binary, returncode) with self.assertRaisesRegex(VERIFY.VerificationError, "expected"): VERIFY.validate_trace_exit(binary, returncode + 1) with self.assertRaisesRegex(VERIFY.VerificationError, "expected None"): VERIFY.validate_trace_exit("example", 0) def test_workspace_binary_discovery_covers_every_current_target(self) -> None: self.assertEqual( set(VERIFY.workspace_binary_targets(VERIFY.workspace_metadata())), { "lazaret", "lazaret-escalation-client", "lazaret-host-coordinator", "lazaret-git-sentry", "lazaret-ignored-workspace", "lazaret-launch-wrapper", "lazaret-runtime-supervisor", "lazaret-sandbox-init", "lazaret-workspace-benchmark", }, ) def test_workspace_binary_names_must_be_unique(self) -> None: metadata = { "workspace_members": ["one", "two"], "packages": [ {"id": "one", "targets": [{"name": "same", "kind": ["bin"]}]}, {"id": "two", "targets": [{"name": "same", "kind": ["bin"]}]}, ], } with self.assertRaisesRegex(VERIFY.VerificationError, "must be unique"): VERIFY.workspace_binary_targets(metadata) def test_binary_artifacts_use_cargo_reported_path(self) -> None: metadata = { "workspace_members": ["one"], "packages": [ {"id": "one", "targets": [{"name": "example", "kind": ["bin"]}]} ], } executable = "/tmp/custom-cargo-target/debug/example" message = ( '{"reason":"compiler-artifact","package_id":"one",' f'"target":{{"name":"example","kind":["bin"]}},"executable":"{executable}"}}' ) self.assertEqual( VERIFY.binary_artifacts(metadata, message), {"example": Path(executable)}, ) def test_tool_versions_are_exactly_pinned(self) -> None: expected = { "rustc": r"^rustc 1\.97\.0 ", "cargo": r"^cargo 1\.97\.0 ", "cargo-nextest": r"^cargo-nextest 0\.9\.140 ", "cargo-deny": r"^cargo-deny 0\.20\.2$", "tlc": r"Version 2\.19 ", "uv": r"^uv 0\.11\.28(?: .*)?$", } self.assertEqual( {program: pattern.pattern for program, _, _, pattern in VERIFY.REQUIRED_TOOL_VERSIONS}, expected, ) def test_entrypoint_uses_dependency_free_offline_uv(self) -> None: entrypoint = (ROOT / "tools/verify").read_text() self.assertIn( '"$(dirname "$0")/materialize-fuser" --test-dependencies', entrypoint ) self.assertIn('"$(dirname "$0")/materialize-runsc"', entrypoint) self.assertIn( "uv run --python '>=3.11' --no-project --offline " "--no-python-downloads --no-cache python3", entrypoint, ) self.assertIn('lazaret_verify.py" "$@"', entrypoint) def test_fuser_materializer_matches_locked_archive(self) -> None: packages = VERIFY.load_toml(ROOT / "Cargo.lock")["package"] [fuser] = [package for package in packages if package["name"] == "fuser"] materializer = (ROOT / "tools/materialize-fuser").read_text() self.assertIn(f"version={fuser['version']}", materializer) self.assertIn(f"crate_checksum={fuser['checksum']}", materializer) self.assertTrue( (ROOT / f"patches/fuser-{fuser['version']}-lock-flags.patch").is_file() ) cargo_config = VERIFY.load_toml(ROOT / ".cargo/config.toml") self.assertEqual( cargo_config["paths"], [f".lazaret/vendor/fuser-{fuser['version']}"], ) def test_runsc_materializer_and_install_identity_are_exact(self) -> None: build_id = "9e711e005f13402e728a0c3ada3fdc67a38918da31f720e5be1006edc4272a51" materializer = (ROOT / "tools/materialize-runsc").read_text() installer = (ROOT / "tools/install-to-host").read_text() runtime = (ROOT / "crates/lazaret-runtime-supervisor/src/sandbox.rs").read_text() self.assertIn( "revision=9f8f6c917885d9ef3d84c17280939c012dea0dfe", materializer ) self.assertIn(f"build_id={build_id}", materializer) self.assertIn(f"runsc_build_id={build_id}", installer) self.assertIn(f'"lazaret-runsc-{build_id}"', runtime) encoded = re.search( r"RUNSC_BUILD_ID_BYTES: \[u8; 32\] = \[(.*?)\];", runtime, re.DOTALL ) self.assertIsNotNone(encoded) self.assertEqual( bytes(int(value.strip()) for value in encoded.group(1).split(",") if value.strip()), bytes.fromhex(build_id), ) self.assertIn("cmp --silent", installer) self.assertTrue( (ROOT / "patches/gvisor-9f8f6c917-directory-handles.patch").is_file() ) self.assertTrue( (ROOT / "patches/gvisor-9f8f6c917-private-ownership.patch").is_file() ) self.assertTrue( ( ROOT / "patches/gvisor-9f8f6c917-per-description-file-handles.patch" ).is_file() ) self.assertTrue( (ROOT / "patches/gvisor-9f8f6c917-hostinet-recv-flags.patch").is_file() ) self.assertTrue( (ROOT / "patches/gvisor-9f8f6c917-gofer-zombie-state.patch").is_file() ) self.assertTrue( (ROOT / "patches/gvisor-9f8f6c917-MODULE.bazel.lock").is_file() ) def test_canonical_build_uses_one_unprivileged_nerdctl_container(self) -> None: build = (ROOT / "tools/build").read_text() containerfile = (ROOT / "tools/container/Containerfile").read_text() installer = (ROOT / "tools/install-to-host").read_text() materializer = (ROOT / "tools/materialize-runsc").read_text() self.assertIn("nerdctl build", build) self.assertIn("nerdctl run --rm", build) self.assertIn("tools/materialize-runsc --build", build) self.assertIn("cargo build", build) self.assertIn("GIT_CEILING_DIRECTORIES=/workspace", build) self.assertNotIn("--privileged", build) self.assertNotIn("docker.sock", build) self.assertIn("rust:1.97.0-bookworm@sha256:", containerfile) self.assertIn("ARG BAZEL_VERSION=8.3.1", containerfile) self.assertIn("tools/build", installer) self.assertNotIn("cargo build", installer) self.assertNotIn("readelf", installer) self.assertIn("bazel build", materializer) self.assertNotIn("make -C", materializer) def test_workspace_tests_use_nextest_and_cargo_retains_doctests(self) -> None: source = (ROOT / "tools/lazaret_verify.py").read_text() expected = [ [ "cargo", "test", "--offline", "--locked", "--manifest-path", ".lazaret/vendor/fuser-0.17.0/Cargo.toml", "lock_request_preserves_kind_flags", ], [ "cargo", "nextest", "run", "--offline", "--workspace", "--all-features", "--locked", "--filterset", ( "not binary(capability-drop-helper) and not binary(acquisition) " "and not binary(state-anchor-lifecycle) and not binary(recovery-process) " "and not binary(runtime-composition) and not binary(escalation) " "and not binary(prepared-worker) and not binary(nofile) " "and not binary(network-runtime)" ), ], [ "cargo", "test", "--offline", "-p", "lazaret-linux", "--test", "acquisition", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-state-anchor", "--test", "state-anchor-lifecycle", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-host-coordinator", "--test", "recovery-process", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-linux", "--test", "capability-drop-helper", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-host-coordinator", "--test", "escalation", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-host-coordinator", "--test", "prepared-worker", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-host-coordinator", "--test", "nofile", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-host-coordinator", "--test", "runtime-composition", "--locked", ], [ "cargo", "test", "--offline", "-p", "lazaret-runtime-supervisor", "--test", "network-runtime", "--locked", ], [ "cargo", "test", "--offline", "--workspace", "--all-features", "--locked", "--doc", ], ] commands = [] original_run = VERIFY.run original_physical = VERIFY.run_physical VERIFY.run = lambda command, cwd=VERIFY.ROOT: commands.append(command) VERIFY.run_physical = lambda command, cwd=VERIFY.ROOT: commands.append(command) try: VERIFY.run_rust_tests() finally: VERIFY.run = original_run VERIFY.run_physical = original_physical self.assertEqual(commands, expected) self.assertIn("exact cargo-nextest 0.9.140", source) self.assertIn("exact uv 0.11.28", source) self.assertIn("Python 3.11 or newer", source) def test_every_harnessless_fixture_is_excluded_and_run_explicitly(self) -> None: harnessless = set() for manifest_path in (ROOT / "crates").glob("*/Cargo.toml"): manifest = VERIFY.load_toml(manifest_path) package = manifest["package"]["name"] for target in manifest.get("test", []): if target.get("harness") is False: harnessless.add((package, target["name"])) nextest_filter = VERIFY.NEXTEST_COMMAND[-1] for _, target in harnessless: self.assertIn(f"not binary({target})", nextest_filter) commands = [] original_run = VERIFY.run original_physical = VERIFY.run_physical VERIFY.run = lambda command, cwd=VERIFY.ROOT: commands.append(command) VERIFY.run_physical = lambda command, cwd=VERIFY.ROOT: commands.append(command) try: VERIFY.run_rust_tests() finally: VERIFY.run = original_run VERIFY.run_physical = original_physical explicit = { (command[command.index("-p") + 1], command[command.index("--test") + 1]) for command in commands if "-p" in command and "--test" in command } self.assertEqual(explicit, harnessless) def test_physical_test_accepts_only_pass_or_explicit_skip(self) -> None: with mock.patch.object( VERIFY.subprocess, "run", return_value=VERIFY.subprocess.CompletedProcess(["physical"], 77), ): VERIFY.run_physical(["physical"]) with mock.patch.object( VERIFY.subprocess, "run", return_value=VERIFY.subprocess.CompletedProcess(["physical"], 1), ): with self.assertRaisesRegex(VERIFY.VerificationError, "physical test exited 1"): VERIFY.run_physical(["physical"]) def test_ignored_workspace_corpus_inventory_has_an_executable_owner(self) -> None: inventory = json.loads( ( ROOT / "tests/fixtures/ignored-workspace-v1/corpus.json" ).read_text() )["cases"] implemented = set(CORPUS.CASE_GROUPS) | {"linked-worktree-index", "git-sentry"} self.assertEqual({case["id"] for case in inventory}, implemented) def test_git_sentry_launch_uses_every_production_authority(self) -> None: self.assertEqual( CORPUS.git_sentry_command( "sentry", "workspace", "common", "hooks", "config" ), ["sentry", "workspace", "common", "hooks", "config"], ) def test_ignored_workspace_prerequisite_absence_is_specific(self) -> None: with mock.patch.object(CORPUS.sys, "platform", "darwin"): self.assertEqual(CORPUS.unavailable_physical_prerequisite(), "Linux is required") with ( mock.patch.object(CORPUS.sys, "platform", "linux"), mock.patch.object(CORPUS.shutil, "which", return_value=None), ): self.assertEqual( CORPUS.unavailable_physical_prerequisite(), "git is unavailable" ) def test_default_verifier_runs_the_complete_ignored_workspace_corpus(self) -> None: binaries = { "lazaret-ignored-workspace": Path("/build/router"), "lazaret-git-sentry": Path("/build/sentry"), } commands = [] with ( mock.patch.object(VERIFY, "build_workspace_binaries", return_value=binaries), mock.patch.object( VERIFY, "run_physical", side_effect=lambda command, cwd=VERIFY.ROOT: commands.append(command), ), ): VERIFY.run_ignored_workspace_corpus({}) [command] = commands self.assertEqual( command[0:2], [VERIFY.sys.executable, str(ROOT / "tools/ignored_workspace_verifier.py")], ) self.assertIn("corpus", command) self.assertIn(str(binaries["lazaret-ignored-workspace"]), command) self.assertIn(str(binaries["lazaret-git-sentry"]), command) self.assertNotIn("--case", command) def test_advisory_refresh_checks_versions_before_mutation(self) -> None: events = [] original_versions = VERIFY.check_tool_versions original_refresh = VERIFY.refresh_advisories VERIFY.check_tool_versions = lambda: events.append("versions") VERIFY.refresh_advisories = lambda: events.append("refresh") try: with mock.patch.object(VERIFY.sys, "argv", ["tools/verify", "--refresh-advisories"]): self.assertEqual(VERIFY.main(), 0) finally: VERIFY.check_tool_versions = original_versions VERIFY.refresh_advisories = original_refresh self.assertEqual(events, ["versions", "refresh"]) def test_model_paths_are_current_and_complete(self) -> None: self.assertEqual( { (module.name, config.name) for module, config in VERIFY.MODEL_CONFIGURATIONS }, { ( "IgnoredWorkspaceHardlinkRouting.tla", "IgnoredWorkspaceHardlinkRouting.cfg", ), ("IgnoredWorkspaceRouting.tla", "IgnoredWorkspaceRouting.cfg"), ("Lazaret.tla", "Lazaret.cfg"), ("Lazaret.tla", "LazaretTerminalLocalControl.cfg"), ("Lazaret.tla", "LazaretTerminalPeerIsolation.cfg"), ("Lazaret.tla", "LazaretTerminalSafety.cfg"), ("PreparedLower.tla", "PreparedLower.cfg"), }, ) for module, config in VERIFY.MODEL_CONFIGURATIONS: self.assertTrue(module.is_file()) self.assertTrue(config.is_file()) def test_model_metadata_directories_are_unique_per_configuration(self) -> None: commands = [] manager = mock.MagicMock() manager.__enter__.return_value = "/verification" manager.__exit__.return_value = False with ( mock.patch.object(VERIFY.shutil, "which", return_value="/usr/bin/tlc"), mock.patch.object(VERIFY, "temporary_directory", return_value=manager), mock.patch.object( VERIFY, "run", side_effect=lambda command, cwd: commands.append(command), ), ): VERIFY.run_models() self.assertEqual( [ ( Path(command[-1]).name, Path(command[command.index("-config") + 1]).name, ) for command in commands ], [ (module.name, config.name) for module, config in VERIFY.MODEL_CONFIGURATIONS ], ) metadata = [command[command.index("-metadir") + 1] for command in commands] self.assertEqual(len(metadata), len(set(metadata))) self.assertEqual( [Path(path).name for path in metadata], [ f"{module.stem}-{config.stem}" for module, config in VERIFY.MODEL_CONFIGURATIONS ], ) def test_lazaret_terminal_configs_retain_common_constants(self) -> None: configs = [ ROOT / "docs/protocol/Lazaret.cfg", ROOT / "docs/protocol/LazaretTerminalSafety.cfg", ROOT / "docs/protocol/LazaretTerminalPeerIsolation.cfg", ROOT / "docs/protocol/LazaretTerminalLocalControl.cfg", ] prefixes = { config.read_text().partition(" TerminalEventPlan <-")[0] for config in configs } self.assertEqual(len(prefixes), 1) def test_lazaret_full_event_safety_retains_canonical_invariants(self) -> None: def invariants(config: Path) -> tuple[str, ...]: body = config.read_text().split("INVARIANTS\n", 1)[1] return tuple(body.split("\n\n", 1)[0].split()) self.assertEqual( invariants(ROOT / "docs/protocol/LazaretTerminalSafety.cfg"), invariants(ROOT / "docs/protocol/Lazaret.cfg"), ) if __name__ == "__main__": unittest.main()