Confine verification tooling to current repository checks
This commit is contained in:
parent
de57e77dc8
commit
c738d92f01
3 changed files with 798 additions and 2926 deletions
7
tests/fixtures/README.md
vendored
7
tests/fixtures/README.md
vendored
|
|
@ -1,7 +0,0 @@
|
|||
# Fixture conventions
|
||||
|
||||
Fixtures are immutable inputs grouped by the verifier that owns their format. Use descriptive lowercase directory names and commit expected outputs beside inputs when exact bytes matter. A fixture directory must contain a `fixture.toml` with `schema`, `case`, and `expected` fields; paths in that file are relative to its directory.
|
||||
|
||||
Positive and negative cases remain separate. Negative cases name the rejected condition and assert the stable error category, not incidental wording. Physical and failure fixtures also record the verification level, durable checkpoint, and exact identity fields under test. Generated, downloaded, host-specific, credential-bearing, and mutable runtime data are not fixtures.
|
||||
|
||||
Tests must copy writable inputs to a test-owned temporary directory and remove it on success or failure. Retained corpus updates are reviewed as input changes; tests never rewrite committed expected output.
|
||||
|
|
@ -1,342 +0,0 @@
|
|||
import copy
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SPEC = importlib.util.spec_from_file_location("lazaret_dev", ROOT / "tools/lazaret_dev.py")
|
||||
DEV = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(DEV)
|
||||
|
||||
|
||||
class RegistryTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.packages = DEV.load_toml(ROOT / "support/work-packages.toml")
|
||||
self.refinements = DEV.load_toml(ROOT / "support/model-refinements.toml")
|
||||
self.external = DEV.load_toml(ROOT / "support/external-tools.toml")
|
||||
self.integrations = DEV.load_toml(ROOT / "support/integration-tests.toml")
|
||||
self.process_safety = DEV.load_toml(ROOT / "support/process-safety.toml")
|
||||
|
||||
def test_committed_registries_are_complete(self):
|
||||
packages = DEV.validate_packages(self.packages)
|
||||
DEV.validate_refinements(self.refinements, {package["id"] for package in packages})
|
||||
DEV.validate_external_tools(self.external)
|
||||
capability_ids = {item["id"] for kind in ("command", "feature") for item in self.external[kind]}
|
||||
DEV.validate_integrations(self.integrations, capability_ids)
|
||||
DEV.validate_process_safety(self.process_safety, self.integrations)
|
||||
|
||||
def test_dependency_cycle_is_rejected(self):
|
||||
packages = copy.deepcopy(self.packages)
|
||||
packages["package"][0]["depends_on"] = ["IP-01"]
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "cycle"):
|
||||
DEV.validate_packages(packages)
|
||||
|
||||
def test_unknown_model_symbol_is_rejected(self):
|
||||
refinements = copy.deepcopy(self.refinements)
|
||||
refinements["coverage"][0]["symbols"].append("UnknownAction")
|
||||
package_ids = {package["id"] for package in DEV.validate_packages(self.packages)}
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "coverage mismatch"):
|
||||
DEV.validate_refinements(refinements, package_ids)
|
||||
|
||||
def test_unknown_probe_policy_is_rejected(self):
|
||||
external = copy.deepcopy(self.external)
|
||||
external["command"][0]["policy"] = "ignore"
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "fail or skip"):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_command_fields_are_strict(self):
|
||||
mutations = (
|
||||
lambda command: command.pop("program"),
|
||||
lambda command: command.update(args=[1]),
|
||||
lambda command: command.update(unexpected=True),
|
||||
)
|
||||
for mutation in mutations:
|
||||
with self.subTest(mutation=mutation):
|
||||
external = copy.deepcopy(self.external)
|
||||
mutation(external["command"][0])
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_each_probe_rejects_missing_fields(self):
|
||||
for index, feature in enumerate(self.external["feature"]):
|
||||
with self.subTest(probe=feature["probe"]):
|
||||
external = copy.deepcopy(self.external)
|
||||
removable = next((field for field in feature if field not in {"id", "probe", "policy", "levels", "diagnostic"}), "diagnostic")
|
||||
external["feature"][index].pop(removable)
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_each_probe_rejects_malformed_fields(self):
|
||||
invalid = {
|
||||
"linux-x86_64": {"id": 1},
|
||||
"kernel-at-least": {"minimum": 6.8},
|
||||
"registered-host-support": {"contract": "unknown"},
|
||||
"user-namespace": {"program": 1},
|
||||
"subordinate-ids": {"minimum_count": "65536"},
|
||||
"open-character-device": {"access": "read-only"},
|
||||
}
|
||||
for index, feature in enumerate(self.external["feature"]):
|
||||
with self.subTest(probe=feature["probe"]):
|
||||
external = copy.deepcopy(self.external)
|
||||
external["feature"][index].update(invalid[feature["probe"]])
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_each_probe_rejects_extraneous_fields(self):
|
||||
for index, feature in enumerate(self.external["feature"]):
|
||||
with self.subTest(probe=feature["probe"]):
|
||||
external = copy.deepcopy(self.external)
|
||||
external["feature"][index]["unexpected"] = True
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "unexpected fields"):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_supported_platform_probe_is_mandatory(self):
|
||||
for capability in ("linux-x86_64", "kernel", "pidfs-host-support", "user-namespaces"):
|
||||
with self.subTest(capability=capability):
|
||||
external = copy.deepcopy(self.external)
|
||||
external["feature"] = [feature for feature in external["feature"] if feature["id"] != capability]
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "external manifest missing"):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_pidfs_contract_fields_are_strict(self):
|
||||
mutations = (
|
||||
lambda external: external["pidfs_host_support"].pop("diagnostic_smoke"),
|
||||
lambda external: external["pidfs_host_support"].update(minimum_kernel=6.9),
|
||||
lambda external: external["pidfs_host_support"].update(unexpected=True),
|
||||
lambda external: external["host_probe"][0].update(steps=["kernel-version-only"]),
|
||||
lambda external: external["host_probe"][0].update(attempts_per_step=True),
|
||||
lambda external: external["host_probe"][0].update(total_timeout_ms=1000),
|
||||
lambda external: external["host_probe"][0].update(step_timeout_ms=4000),
|
||||
lambda external: external["kernel_qualification"][0].pop("source_commit"),
|
||||
lambda external: external["kernel_qualification"][0].update(build_image_sha256="bad"),
|
||||
lambda external: external["kernel_qualification"][0].update(build_image_sha256=1),
|
||||
lambda external: external["kernel_qualification"][0].update(id=[]),
|
||||
lambda external: external["kernel_qualification"][0].update(unexpected=True),
|
||||
)
|
||||
for mutation in mutations:
|
||||
with self.subTest(mutation=mutation):
|
||||
external = copy.deepcopy(self.external)
|
||||
mutation(external)
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_linux_68_is_refused_and_69_meets_only_the_floor(self):
|
||||
kernel = next(feature for feature in self.external["feature"] if feature["id"] == "kernel")
|
||||
with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object(
|
||||
DEV.platform, "release", return_value="6.8.99"
|
||||
):
|
||||
self.assertEqual(DEV.probe_feature(kernel), (False, "6.8.99"))
|
||||
with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object(
|
||||
DEV.platform, "release", return_value="6.9.0"
|
||||
):
|
||||
self.assertEqual(DEV.probe_feature(kernel), (True, "6.9.0"))
|
||||
with mock.patch.object(DEV.platform, "system", return_value="Linux"), mock.patch.object(
|
||||
DEV.platform, "release", return_value="6.12.38+deb13-amd64"
|
||||
):
|
||||
self.assertEqual(DEV.probe_feature(kernel), (True, "6.12.38+deb13-amd64"))
|
||||
pidfs = next(feature for feature in self.external["feature"] if feature["id"] == "pidfs-host-support")
|
||||
self.assertEqual(DEV.probe_feature(pidfs), (False, "registered runner pending IP-03"))
|
||||
|
||||
def test_pidfs_smoke_failure_is_a_hard_runtime_refusal(self):
|
||||
contract = self.external["pidfs_host_support"]
|
||||
self.assertTrue(contract["diagnostic_smoke"].startswith("LZ-HOST-PIDFS-SMOKE:"))
|
||||
self.assertEqual(contract["runtime_failure_policy"], "fail-before-generation")
|
||||
|
||||
def test_mandatory_capabilities_cannot_be_remapped(self):
|
||||
for capability in ("linux-x86_64", "kernel", "pidfs-host-support", "user-namespaces"):
|
||||
with self.subTest(capability=capability):
|
||||
external = copy.deepcopy(self.external)
|
||||
feature = next(item for item in external["feature"] if item["id"] == capability)
|
||||
feature.update(policy="fail")
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "mandatory feature contract drift"):
|
||||
DEV.validate_external_tools(external)
|
||||
external = copy.deepcopy(self.external)
|
||||
pidfs = next(item for item in external["feature"] if item["id"] == "pidfs-host-support")
|
||||
pidfs.pop("contract")
|
||||
pidfs["probe"] = "linux-x86_64"
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "mandatory feature contract drift"):
|
||||
DEV.validate_external_tools(external)
|
||||
external = copy.deepcopy(self.external)
|
||||
rustc = next(item for item in external["command"] if item["id"] == "rustc")
|
||||
rustc["levels"] = ["V1"]
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "mandatory command contract drift"):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_unqualified_vendor_build_is_rejected(self):
|
||||
external = copy.deepcopy(self.external)
|
||||
vendor = copy.deepcopy(external["kernel_qualification"][0])
|
||||
vendor.update(
|
||||
id="vendor-linux-6.9.0-x86_64",
|
||||
kind="vendor",
|
||||
vendor="Example Vendor",
|
||||
vendor_patchset_sha256="0" * 64,
|
||||
status="unqualified",
|
||||
)
|
||||
external["kernel_qualification"].append(vendor)
|
||||
external["pidfs_host_support"]["qualification_ids"].append(vendor["id"])
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "not qualified"):
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_qualified_vendor_release_string_is_representable(self):
|
||||
external = copy.deepcopy(self.external)
|
||||
vendor = copy.deepcopy(external["kernel_qualification"][0])
|
||||
vendor.update(
|
||||
id="vendor-linux-6.12-x86_64",
|
||||
kind="vendor",
|
||||
release="6.12.38-1-amd64",
|
||||
vendor="Example Vendor",
|
||||
vendor_patchset_sha256="0" * 64,
|
||||
)
|
||||
external["kernel_qualification"].append(vendor)
|
||||
external["pidfs_host_support"]["qualification_ids"].append(vendor["id"])
|
||||
DEV.validate_external_tools(external)
|
||||
|
||||
def test_process_safety_registration_is_strict(self):
|
||||
mutations = (
|
||||
lambda process: process["policy"].pop("dynamic_scenario"),
|
||||
lambda process: process["policy"].update(production_roots=["crates/lazaret-core"]),
|
||||
lambda process: process["policy"].update(lint_allow_paths=["crates/foreign.rs"]),
|
||||
lambda process: process["policy"].update(trace_expression="read"),
|
||||
lambda process: process["policy"].update(forbidden_syscall_patterns=["$^"]),
|
||||
lambda process: process["policy"].update(unexpected=True),
|
||||
lambda process: process["runtime_dependency"][0].update(features=["z", "a"]),
|
||||
lambda process: process["runtime_dependency"][0].update(checksum=1),
|
||||
lambda process: process["trace_case"][0].update(initial_execve_count=True),
|
||||
)
|
||||
for mutation in mutations:
|
||||
with self.subTest(mutation=mutation):
|
||||
process = copy.deepcopy(self.process_safety)
|
||||
mutation(process)
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_process_safety(process, self.integrations)
|
||||
integrations = copy.deepcopy(self.integrations)
|
||||
integrations["scenarios"] = []
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "dynamic syscall scenario"):
|
||||
DEV.validate_process_safety(self.process_safety, integrations)
|
||||
|
||||
def test_process_lint_exemption_is_detected(self):
|
||||
self.assertTrue(DEV.has_process_lint_exemption("#![allow(clippy::disallowed_types)]\n"))
|
||||
self.assertTrue(DEV.has_process_lint_exemption("#[expect(clippy::{disallowed_methods})]\n"))
|
||||
|
||||
def test_workspace_process_lints_are_mandatory(self):
|
||||
workspace = DEV.load_toml(ROOT / "Cargo.toml")
|
||||
DEV.validate_workspace_process_lints(workspace)
|
||||
for group, lint in (("rust", "unsafe_code"), ("clippy", "disallowed_methods"), ("clippy", "disallowed_types")):
|
||||
with self.subTest(lint=lint):
|
||||
changed = copy.deepcopy(workspace)
|
||||
changed["workspace"]["lints"][group].pop(lint)
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_workspace_process_lints(changed)
|
||||
|
||||
def test_every_workspace_member_inherits_process_lints(self):
|
||||
workspace = DEV.load_toml(ROOT / "Cargo.toml")["workspace"]
|
||||
for member in workspace["members"]:
|
||||
with self.subTest(member=member):
|
||||
manifest = DEV.load_toml(ROOT / member / "Cargo.toml")
|
||||
DEV.validate_member_process_lints(manifest, member)
|
||||
changed = copy.deepcopy(manifest)
|
||||
changed["lints"] = {"workspace": False}
|
||||
with self.assertRaises(DEV.RegistryError):
|
||||
DEV.validate_member_process_lints(changed, member)
|
||||
|
||||
def test_compiler_resolves_grouped_process_import_alias(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
(root / "src").mkdir()
|
||||
(root / "Cargo.toml").write_text(
|
||||
'[package]\nname = "lint-probe"\nversion = "0.0.0"\nedition = "2024"\n'
|
||||
"[lints.clippy]\ndisallowed_types = \"deny\"\n"
|
||||
)
|
||||
(root / ".clippy.toml").write_text(DEV.clippy_config_view(self.process_safety))
|
||||
(root / "src/main.rs").write_text(
|
||||
'use std::{process::Command as C};\nfn main() { let _ = C::new("true"); }\n'
|
||||
)
|
||||
result = subprocess.run(
|
||||
["cargo", "clippy", "--offline", "--", "-D", "warnings"],
|
||||
cwd=root,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
check=False,
|
||||
)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("disallowed type", result.stdout)
|
||||
|
||||
def test_process_syscall_trace_refuses_bypass(self):
|
||||
policy = self.process_safety["policy"]
|
||||
trace_case = self.process_safety["trace_case"][0]
|
||||
DEV.validate_process_trace(['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0'], policy, trace_case)
|
||||
for syscall in ("clone", "execveat", "wait4", "waitid"):
|
||||
with self.subTest(syscall=syscall), self.assertRaisesRegex(DEV.RegistryError, "forbidden"):
|
||||
DEV.validate_process_trace(
|
||||
['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0', f"{syscall}(0) = 1"],
|
||||
policy,
|
||||
trace_case,
|
||||
)
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "SIGCHLD"):
|
||||
DEV.validate_process_trace(
|
||||
['execve("/tmp/lazaret", ["lazaret"], 0x0) = 0', "rt_sigaction(SIGCHLD, {}, NULL, 8) = 0"],
|
||||
policy,
|
||||
trace_case,
|
||||
)
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "execve surface count"):
|
||||
DEV.validate_process_trace(
|
||||
['execve("/tmp/lazaret", [], 0x0) = 0', 'execve("/tmp/other", [], 0x0) = 0'],
|
||||
policy,
|
||||
trace_case,
|
||||
)
|
||||
|
||||
def test_generated_host_view_covers_authoritative_fields(self):
|
||||
baseline = DEV.host_contract_view(self.external, self.integrations, self.process_safety)
|
||||
external = copy.deepcopy(self.external)
|
||||
external["kernel_qualification"][0]["probe_binary_sha256"] = "0" * 64
|
||||
self.assertNotEqual(DEV.host_contract_view(external, self.integrations, self.process_safety), baseline)
|
||||
process = copy.deepcopy(self.process_safety)
|
||||
process["runtime_dependency"][0]["features"] = ["changed"]
|
||||
self.assertNotEqual(DEV.host_contract_view(self.external, self.integrations, process), baseline)
|
||||
|
||||
def test_integration_manifest_rejects_top_level_extras(self):
|
||||
integrations = copy.deepcopy(self.integrations)
|
||||
integrations["unexpected"] = True
|
||||
capability_ids = {item["id"] for kind in ("command", "feature") for item in self.external[kind]}
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "unexpected fields"):
|
||||
DEV.validate_integrations(integrations, capability_ids)
|
||||
|
||||
def test_missing_skip_capability_skips_scenario(self):
|
||||
integrations = {
|
||||
"scenarios": [{"name": "fuse", "level": "V2", "command": ["false"], "capabilities": ["fuse-device"]}]
|
||||
}
|
||||
with mock.patch.object(DEV, "probe", return_value=(set(), {"fuse-device"})):
|
||||
with contextlib.redirect_stdout(io.StringIO()) as output:
|
||||
DEV.run_integrations("V2", self.external, integrations)
|
||||
self.assertIn("SKIP fuse", output.getvalue())
|
||||
|
||||
def test_missing_fail_capability_rejects_scenario(self):
|
||||
integrations = {
|
||||
"scenarios": [{"name": "compiler", "level": "V2", "command": ["false"], "capabilities": ["rustc"]}]
|
||||
}
|
||||
with mock.patch.object(DEV, "probe", return_value=(set(), set())):
|
||||
with self.assertRaisesRegex(DEV.RegistryError, "undeclared capability failure"):
|
||||
DEV.run_integrations("V2", self.external, integrations)
|
||||
|
||||
def test_subordinate_range_accepts_name_or_numeric_uid(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "subids"
|
||||
path.write_text("developer:not-a-number:65536\n1000:524288:65536\n")
|
||||
self.assertEqual(DEV.subordinate_id_range(path, {"developer", "1000"}, 65536), 524288)
|
||||
|
||||
def test_subordinate_range_rejects_short_or_malformed_rows(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "subids"
|
||||
path.write_text("developer:524288:65535\ndeveloper:bad:65536\ndeveloper:0:65536\n")
|
||||
self.assertIsNone(DEV.subordinate_id_range(path, {"developer", "1000"}, 65536))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
275
tests/tooling/test_verify.py
Normal file
275
tests/tooling/test_verify.py
Normal file
|
|
@ -0,0 +1,275 @@
|
|||
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_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_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-network-sidecar",
|
||||
"lazaret-runtime-supervisor",
|
||||
},
|
||||
)
|
||||
|
||||
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\.90\.0 ",
|
||||
"cargo": r"^cargo 1\.90\.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",
|
||||
],
|
||||
[
|
||||
"cargo",
|
||||
"test",
|
||||
"--offline",
|
||||
"--workspace",
|
||||
"--all-features",
|
||||
"--locked",
|
||||
"--doc",
|
||||
],
|
||||
]
|
||||
commands = []
|
||||
original_run = VERIFY.run
|
||||
VERIFY.run = lambda command, cwd=VERIFY.ROOT: commands.append(command)
|
||||
try:
|
||||
VERIFY.run_rust_tests()
|
||||
finally:
|
||||
VERIFY.run = original_run
|
||||
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_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()
|
||||
Loading…
Reference in a new issue