lazaret/tests/tooling/test_verify.py

from __future__ import annotations

import copy
import importlib.machinery
import importlib.util
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)


class VerificationTests(unittest.TestCase):
    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_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_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:
        with self.assertRaisesRegex(VERIFY.VerificationError, "exercised path exited 1"):
            VERIFY.validate_trace_exit("example", 1)

    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(
            "uv run --python '>=3.11' --no-project --offline "
            "--no-python-downloads --no-cache python3",
            entrypoint,
        )
        self.assertIn('lazaret_verify.py" "$@"', entrypoint)

    def test_workspace_tests_use_nextest_and_cargo_retains_doctests(self) -> None:
        source = (ROOT / "tools/lazaret_verify.py").read_text()
        expected = [
            [
                "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_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()