lazaret/tests/tooling/test_devtool.py

import copy
import contextlib
import importlib.util
import io
from pathlib import Path
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")

    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)

    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},
            "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", "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_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()