mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-09-21 08:50:42 +02:00
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""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)
|