85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
"""Focused unit coverage for the privileged runtime publication seam."""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from tht import runtime_config_lease_io as lease_io
|
|
|
|
|
|
def _manifest() -> dict:
|
|
return {
|
|
"version": 1, "workspace_id": "abc", "workspace_revision": "a" * 40,
|
|
"descriptor_git_blob": "b" * 40, "descriptor_sha256": "c" * 64,
|
|
"descriptor_dev": "1", "descriptor_ino": "2", "config_sha256": "d" * 64,
|
|
"config_dwh_binding": {
|
|
"workspace_id": "abc", "config_fingerprint": "e", "input_fingerprint": "f"
|
|
},
|
|
"config_dev": "1", "config_ino": "3", "config_size": "4",
|
|
"config_mode": "400", "config_uid": str(os.getuid()), "config_nlink": "1",
|
|
"directory_identities": [{"path": "/", "dev": "1", "ino": "1", "mode": "755", "uid": "0"}],
|
|
}
|
|
|
|
|
|
def test_strict_manifest_rejects_unknown_or_missing_fields():
|
|
value = _manifest()
|
|
assert lease_io.strict_manifest(value) is value
|
|
with pytest.raises(RuntimeError):
|
|
lease_io.strict_manifest({**value, "unexpected": True})
|
|
missing = dict(value)
|
|
del missing["config_sha256"]
|
|
with pytest.raises(RuntimeError):
|
|
lease_io.strict_manifest(missing)
|
|
|
|
|
|
def test_private_alias_is_darwin_only(monkeypatch):
|
|
monkeypatch.setattr(lease_io.sys, "platform", "linux")
|
|
assert lease_io._canonical_root("/tmp/runtime") == "/tmp/runtime"
|
|
assert lease_io._canonical_root("/var/lib/runtime") == "/var/lib/runtime"
|
|
monkeypatch.setattr(lease_io.sys, "platform", "darwin")
|
|
assert lease_io._canonical_root("/tmp/runtime") == "/private/tmp/runtime"
|
|
assert lease_io._canonical_root("/var/lib/runtime") == "/private/var/lib/runtime"
|
|
|
|
|
|
def test_read_all_enforces_bound(tmp_path):
|
|
path = tmp_path / "large"
|
|
path.write_bytes(b"0123456789")
|
|
fd = os.open(path, os.O_RDONLY)
|
|
try:
|
|
with pytest.raises(RuntimeError, match="too large"):
|
|
lease_io.read_all(fd, limit=4)
|
|
with pytest.raises(RuntimeError, match="too large"):
|
|
lease_io.read_all(fd, limit=0)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
@pytest.mark.parametrize("failure_stage", ["config-parent", "manifest-parent"])
|
|
def test_retry_reasserts_parent_durability_before_success(tmp_path, monkeypatch, failure_stage):
|
|
events: list[str] = []
|
|
failed = False
|
|
|
|
def fsync(fd: int, stage: str) -> None:
|
|
nonlocal failed
|
|
events.append(stage)
|
|
if stage == failure_stage and not failed:
|
|
failed = True
|
|
raise RuntimeError("injected fsync failure")
|
|
|
|
monkeypatch.setattr(lease_io, "publication_fsync", fsync)
|
|
inp = {
|
|
"data_root": str(tmp_path / "data"), "workspace_id": "abc",
|
|
"workspace_revision": "a" * 40, "config_hex": b"config".hex(),
|
|
"manifest_base": {
|
|
"workspace_id": "abc", "workspace_revision": "a" * 40,
|
|
"descriptor_git_blob": "b" * 40, "descriptor_sha256": "c" * 64,
|
|
"descriptor_dev": "1", "descriptor_ino": "2",
|
|
"config_dwh_binding": {"workspace_id": "abc", "config_fingerprint": "e", "input_fingerprint": "f"},
|
|
},
|
|
}
|
|
with pytest.raises(RuntimeError, match="injected"):
|
|
lease_io.publish(inp)
|
|
events.clear()
|
|
lease_io.publish(inp)
|
|
assert events.index("config-parent") < events.index("manifest-parent")
|