lazaret/tests/tooling/test_verify.py

from __future__ import annotations

import copy
import importlib.machinery
import importlib.util
import json
from pathlib import Path
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-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(
            "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_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(recovery-process) and not binary(runtime-composition) "
                    "and not binary(network-runtime)"
                ),
            ],
            [
                "cargo",
                "test",
                "--offline",
                "-p",
                "lazaret-linux",
                "--test",
                "acquisition",
                "--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",
                "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_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 for module, _ in VERIFY.MODEL_CONFIGURATIONS},
            {"Lazaret.tla", "PreparedLower.tla"},
        )
        for module, config in VERIFY.MODEL_CONFIGURATIONS:
            self.assertTrue(module.is_file())
            self.assertTrue(config.is_file())


if __name__ == "__main__":
    unittest.main()