"""Deterministic failure and system-probe contracts for distributed admission.""" from dataclasses import replace import ctypes import json from types import SimpleNamespace import pytest from obliteratus.distributed import preflight from obliteratus.distributed.config import DistributedPreflightConfig from obliteratus.distributed.contracts import ContractError, RuntimeContractError from tests.test_distributed_launcher import _profile pytestmark = pytest.mark.cpu @pytest.mark.parametrize( ("section", "key", "value", "message"), [ ("run", "world_size", True, "integer"), ("run", "world_size", 1, "between"), ("run", "run_id", "", "bounded non-empty"), ("run", "run_id", "bad", "invalid format"), ("software", "python", "\ud800", "valid UTF-8"), ("source", "path", "relative", "absolute local"), ("network", "allowed_master_cidrs", [], "non-empty bounded"), ("network", "allowed_master_cidrs", ["bad"], "invalid network"), ("network", "allowed_master_cidrs", ["8.8.8.0/24"], "private networks"), ("network", "allowed_master_cidrs", ["10.0.0.0/8"] * 2, "duplicates"), ("topology", "dimension_divisors", [], "non-empty bounded"), ("execution", "allowed_environment_keys", ["PATH", "PATH"], "unique bounded"), ("execution", "allowed_environment_keys", ["HF_TOKEN"], "secret or proxy"), ], ) def test_profile_rejects_invalid_field_values(tmp_path, section, key, value, message): _profile(tmp_path) path = tmp_path / "profile.json" payload = json.loads(path.read_text()) payload[section][key] = value path.write_text(json.dumps(payload)) with pytest.raises(ContractError, match=message): DistributedPreflightConfig.from_file(path) @pytest.mark.parametrize( ("raw", "message"), [ (b"", "regular file"), (b"\xff", "strict UTF-8 JSON"), (b"{", "strict UTF-8 JSON"), (b"[]", "must be an object"), (b'{"schema_version":1,"schema_version":1}', "duplicate"), (b"{}", "required profile section"), ], ) def test_profile_rejects_invalid_serialization(tmp_path, raw, message): path = tmp_path / "profile.json" path.write_bytes(raw) with pytest.raises(ContractError, match=message): DistributedPreflightConfig.from_file(path) @pytest.mark.parametrize( ("change", "message"), [ ({"rendezvous_id": "1" * 32}, "distinct"), ({"local_world_size": 3}, "divisible"), ({"tensor_parallel_size": 4}, "equal world_size"), ({"max_source_bytes": 1}, "cannot exceed"), ({"coordinator_rank": 2}, "smaller than"), ({"device_kind": "mps"}, "cuda or cpu"), ({"evidence_tier": "protocol_cpu"}, "does not match"), ({"local_files_only": False}, "remain true"), ], ) def test_direct_profile_replacement_rechecks_admission(tmp_path, change, message): with pytest.raises(ContractError, match=message): replace(_profile(tmp_path), **change).validate() @pytest.mark.parametrize("kind", ["missing", "symlink", "directory"]) def test_profile_requires_regular_file(tmp_path, kind): target = tmp_path / "target" if kind == "symlink": target.symlink_to(tmp_path / "absent") elif kind == "directory": target.mkdir() with pytest.raises(ContractError, match="regular"): DistributedPreflightConfig.from_file(target) @pytest.mark.parametrize("kind", ["empty", "oversized", "invalid_utf8", "missing"]) def test_checkout_identity_file_is_bounded_and_utf8(tmp_path, kind): path = tmp_path / "HEAD" if kind != "missing": path.write_bytes({"empty": b"", "oversized": b"a" * 4097, "invalid_utf8": b"\xff"}[kind]) with pytest.raises(ContractError, match="identity"): preflight._bounded_text_file(path) @pytest.mark.parametrize( "head", ["garbage", "ref: outside", "ref: refs/../secret", "ref: refs/heads/bad\\name"] ) def test_checkout_rejects_unsafe_head_references(tmp_path, head): git = tmp_path / ".git" git.mkdir() (git / "HEAD").write_text(head) with pytest.raises(ContractError, match="invalid format"): preflight.checkout_commit(tmp_path) def test_checkout_resolves_worktree_common_packed_reference(tmp_path): worktree = tmp_path / "worktree" worktree.mkdir() metadata = tmp_path / "metadata" metadata.mkdir() common = tmp_path / "common" common.mkdir() (worktree / ".git").write_text("gitdir: ../metadata\n") (metadata / "HEAD").write_text("ref: refs/heads/main\n") (metadata / "commondir").write_text("../common\n") commit = "a" * 40 (common / "packed-refs").write_text(f"# packed refs\n{commit} refs/heads/main\n^{commit}\n") assert preflight.checkout_commit(worktree) == commit (common / "packed-refs").write_text(f"{commit} refs/heads/other\n") with pytest.raises(ContractError, match="reference is unavailable"): preflight.checkout_commit(worktree) @pytest.mark.parametrize("mode", ["init_failure", "version_failure", "unavailable", "success"]) def test_driver_version_probe_handles_library_results(monkeypatch, mode): def version(pointer): ctypes.cast(pointer, ctypes.POINTER(ctypes.c_int))[0] = 12040 return int(mode == "version_failure") def library(name): assert name == "libcuda.so.1" if mode == "unavailable": raise OSError("no driver installed") return SimpleNamespace( cuInit=lambda flags: int(mode == "init_failure"), cuDriverGetVersion=version ) monkeypatch.setattr(preflight.ctypes, "CDLL", library) assert preflight._cuda_driver_version() == ("12040" if mode == "success" else "unavailable") @pytest.mark.parametrize( "version, expected", [((2, 28, 3), "2.28.3"), (22803, "22803"), (None, "unavailable")] ) def test_nccl_probe_normalizes_version_and_missing_backend(monkeypatch, version, expected): def probe(): if version is None: raise RuntimeError("backend unavailable") return version monkeypatch.setattr(preflight.torch.cuda.nccl, "version", probe) assert preflight._nccl_version() == expected @pytest.fixture def system_probe_inputs(tmp_path, monkeypatch): config = _profile(tmp_path) source = preflight.SourceIdentity("a" * 64, "b" * 64, "c" * 64, 3, 256) monkeypatch.setattr(preflight, "inspect_source", lambda *args, **kwargs: source) monkeypatch.setattr(preflight, "validate_network_interface", lambda *args, **kwargs: None) monkeypatch.setattr(preflight, "_host_memory", lambda: (16000, 8000)) monkeypatch.setattr(preflight.shutil, "disk_usage", lambda path: SimpleNamespace(free=32000)) monkeypatch.setattr(preflight, "_nccl_version", lambda: "2.28.3") monkeypatch.setattr(preflight, "_cuda_driver_version", lambda: "12040") monkeypatch.setattr(preflight, "checkout_commit", lambda: "d" * 40) monkeypatch.setattr(preflight, "checkout_code_digest", lambda: "e" * 64) monkeypatch.setattr(preflight, "storage_mount_digest", lambda path: "f" * 64) monkeypatch.setattr(preflight.platform, "node", lambda: "test-host") monkeypatch.setattr(preflight.platform, "machine", lambda: "x86_64") return config, SimpleNamespace(rank=0, local_rank=1), source @pytest.mark.parametrize("kind", ["cpu", "cuda"]) def test_system_probe_records_measured_resources(system_probe_inputs, monkeypatch, kind): config, launch, source = system_probe_inputs selected = [] monkeypatch.setattr(preflight.torch.cuda, "is_available", lambda: True) monkeypatch.setattr(preflight.torch.cuda, "set_device", selected.append) monkeypatch.setattr( preflight.torch.cuda, "get_device_properties", lambda rank: SimpleNamespace(uuid="GPU-test", name="Test GPU", major=8, minor=6), ) monkeypatch.setattr(preflight.torch.cuda, "mem_get_info", lambda rank: (4000, 12000)) result = preflight.SystemProbes().collect(replace(config, device_kind=kind), launch) assert result.source == source assert ( result.total_host_memory_bytes, result.free_host_memory_bytes, result.free_staging_bytes, ) == (16000, 8000, 32000) assert result.commit_sha == "d" * 40 assert result.code_digest == "e" * 64 assert result.storage_identity == "f" * 64 assert dict(result.software_versions)["driver"] == "12040" if kind == "cuda": assert selected == [1] assert (result.device_identity, result.device_name, result.compute_capability) == ( "GPU-test", "Test GPU", "8.6", ) assert (result.total_device_memory_bytes, result.free_device_memory_bytes) == (12000, 4000) else: assert selected == [] assert result.device_identity == "cpu:test-host:x86_64:1" assert (result.total_device_memory_bytes, result.free_device_memory_bytes) == (16000, 8000) @pytest.mark.parametrize("failure", ["unavailable", "no_uuid", "checkout", "staging"]) def test_system_probe_fails_closed_on_missing_identity(system_probe_inputs, monkeypatch, failure): config, launch, _ = system_probe_inputs monkeypatch.setattr(preflight.torch.cuda, "is_available", lambda: failure != "unavailable") monkeypatch.setattr(preflight.torch.cuda, "set_device", lambda rank: None) monkeypatch.setattr( preflight.torch.cuda, "get_device_properties", lambda rank: SimpleNamespace() ) if failure == "checkout": config = replace(config, device_kind="cpu") def missing_commit(): raise ContractError("unavailable") monkeypatch.setattr(preflight, "checkout_commit", missing_commit) if failure == "staging": config = replace(config, staging_path=config.staging_path / "missing") with pytest.raises(RuntimeContractError) as error: preflight.SystemProbes().collect(config, launch) assert error.value.code == ( "LMS_STORAGE_PROFILE_MISMATCH" if failure == "staging" else "LMS_RUNTIME_PROFILE_MISMATCH" ) @pytest.mark.parametrize( "kind, code", [ ("writable", "LMS_SOURCE_BOUNDARY_VIOLATION"), ("hardlink", "LMS_SOURCE_BOUNDARY_VIOLATION"), ("oversized", "LMS_RESOURCE_ADMISSION_DENIED"), ("expired", "LMS_STAGE_TIMEOUT"), ("missing", "LMS_SOURCE_BOUNDARY_VIOLATION"), ("replaced", "LMS_SOURCE_CHANGED"), ], ) def test_source_hash_rejects_unsafe_or_changed_file(tmp_path, monkeypatch, kind, code): path = tmp_path / "source.json" path.write_bytes(b"original") path.chmod(0o444) if kind == "writable": path.chmod(0o644) elif kind == "hardlink": (tmp_path / "alias.json").hardlink_to(path) elif kind == "missing": path.unlink() elif kind == "replaced": original_read = preflight.os.read replaced = False def replace_after_read(descriptor, size): nonlocal replaced result = original_read(descriptor, size) if not replaced: replacement = tmp_path / "replacement.json" replacement.write_bytes(b"modified") replacement.chmod(0o444) replacement.replace(path) replaced = True return result monkeypatch.setattr(preflight.os, "read", replace_after_read) monkeypatch.setattr(preflight.time, "monotonic", lambda: 10.0) with pytest.raises(RuntimeContractError) as error: preflight._hash_file( path, deadline=9.0 if kind == "expired" else 11.0, max_bytes=1 if kind == "oversized" else 100, ) assert error.value.code == code def test_host_memory_converts_system_page_counts_to_bytes(monkeypatch): values = {"SC_PAGE_SIZE": 4096, "SC_PHYS_PAGES": 1000, "SC_AVPHYS_PAGES": 250} monkeypatch.setattr(preflight.os, "sysconf", values.__getitem__) assert preflight._host_memory() == (4_096_000, 1_024_000)