617 lines
22 KiB
Python
617 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPT = REPO_ROOT / "scripts/reconcile_mediarr_validation_policy.py"
|
|
REGISTRY_SOURCE = REPO_ROOT / "backend/app/services/deployment_target_registry.py"
|
|
SPEC = importlib.util.spec_from_file_location("mediarr_validation_policy", SCRIPT)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
policy = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(policy)
|
|
|
|
PUSH_CONTEXTS = ["CI / gates (push)", "Managed validation / full (push)"]
|
|
ACTIVE_STATUSES = (
|
|
"pending",
|
|
"running",
|
|
"awaiting_validation",
|
|
"validation_passed",
|
|
"retry_wait",
|
|
"deploying",
|
|
"future_unrecognized_state",
|
|
)
|
|
TERMINAL_STATUSES = (
|
|
"deployed",
|
|
"non_forward",
|
|
"failed",
|
|
"validation_failed",
|
|
"validation_timeout",
|
|
"superseded",
|
|
"rejected",
|
|
)
|
|
|
|
|
|
def _projects(validation: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
project = copy.deepcopy(policy.TARGET_IDENTITY)
|
|
project["validation"] = copy.deepcopy(
|
|
policy.LEGACY_POLICY if validation is None else validation
|
|
)
|
|
return {policy.REPOSITORY: project}
|
|
|
|
|
|
def _extract_registry_compatibility():
|
|
"""Load only the inline predicate; importing the registry module has host side effects."""
|
|
registry_tree = ast.parse(
|
|
REGISTRY_SOURCE.read_text(encoding="utf-8"), filename=str(REGISTRY_SOURCE)
|
|
)
|
|
register_script = next(
|
|
(
|
|
node
|
|
for node in registry_tree.body
|
|
if isinstance(node, ast.Assign)
|
|
and any(
|
|
isinstance(target, ast.Name) and target.id == "_REGISTER_SCRIPT"
|
|
for target in node.targets
|
|
)
|
|
),
|
|
None,
|
|
)
|
|
assert register_script is not None, "registry must keep its fixed registration script inline"
|
|
source = ast.literal_eval(register_script.value)
|
|
tree = ast.parse(source, filename=f"{REGISTRY_SOURCE}:_REGISTER_SCRIPT")
|
|
function = next(
|
|
(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
and node.name == "_validation_compatible"
|
|
),
|
|
None,
|
|
)
|
|
assert function is not None, "inline registration script must define _validation_compatible"
|
|
|
|
# The predicate may use module-level literal policy constants. Extract only literals and the
|
|
# function, deliberately excluding imports, filesystem setup, controller state, and CLI code.
|
|
namespace: dict[str, Any] = {"__builtins__": __builtins__}
|
|
body: list[ast.stmt] = []
|
|
for node in tree.body:
|
|
if isinstance(node, ast.ImportFrom) and node.module == "__future__":
|
|
body.append(node)
|
|
elif isinstance(node, ast.Assign):
|
|
try:
|
|
ast.literal_eval(node.value)
|
|
except (ValueError, TypeError, SyntaxError):
|
|
continue
|
|
if all(isinstance(target, ast.Name) for target in node.targets):
|
|
body.append(node)
|
|
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
|
|
try:
|
|
ast.literal_eval(node.value)
|
|
except (ValueError, TypeError, SyntaxError, AttributeError):
|
|
continue
|
|
body.append(node)
|
|
body.append(function)
|
|
module = ast.fix_missing_locations(ast.Module(body=body, type_ignores=[]))
|
|
exec(compile(module, str(REGISTRY_SOURCE), "exec"), namespace) # noqa: S102
|
|
return namespace["_validation_compatible"]
|
|
|
|
|
|
def _registry_bytes(projects: dict[str, Any]) -> bytes:
|
|
return (json.dumps(projects, indent=2, sort_keys=True) + "\n").encode("utf-8")
|
|
|
|
|
|
def _write_json(path: Path, value: Any, *, mode: int = 0o600) -> bytes:
|
|
payload = _registry_bytes(value) if path.name == "projects.json" else (
|
|
json.dumps(value, sort_keys=True) + "\n"
|
|
).encode("utf-8")
|
|
path.write_bytes(payload)
|
|
if os.name == "posix":
|
|
path.chmod(mode)
|
|
return payload
|
|
|
|
|
|
def _prepare_filesystem(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
validation: dict[str, Any] | None = None,
|
|
inbox_records: dict[str, dict[str, Any]] | None = None,
|
|
) -> dict[str, Path | bytes]:
|
|
runtime_root = tmp_path / "runner-root"
|
|
config_root = tmp_path / "protected-config"
|
|
state_root = runtime_root / "state"
|
|
inbox = state_root / "webhook-inbox"
|
|
lock_root = state_root / "locks"
|
|
backup_root = state_root / "registry-backups"
|
|
registry = config_root / "projects.json"
|
|
project_lock = lock_root / "mediarr.lock"
|
|
global_lock = runtime_root / "deploy.lock"
|
|
receipt_root = state_root / "receipts"
|
|
|
|
for directory in (
|
|
runtime_root,
|
|
config_root,
|
|
state_root,
|
|
inbox,
|
|
lock_root,
|
|
receipt_root,
|
|
backup_root,
|
|
):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
if os.name == "posix":
|
|
directory.chmod(0o700)
|
|
|
|
original = _write_json(registry, _projects(validation))
|
|
for lock in (project_lock, global_lock):
|
|
lock.touch(mode=0o600)
|
|
if os.name == "posix":
|
|
lock.chmod(0o600)
|
|
|
|
# A completed delivery is terminal history, not active work. It must remain byte-for-byte.
|
|
records = inbox_records
|
|
if records is None:
|
|
records = {
|
|
f"terminal-{status}.json": {
|
|
"repository": policy.REPOSITORY,
|
|
"status": status,
|
|
"receipt": f"fixture-{status}-receipt",
|
|
}
|
|
for status in TERMINAL_STATUSES
|
|
}
|
|
records["other-repository.json"] = {
|
|
"repository": "jens/another-project",
|
|
"status": "running",
|
|
"receipt": "fixture-other-receipt",
|
|
}
|
|
for filename, record in records.items():
|
|
_write_json(inbox / filename, record)
|
|
receipt = receipt_root / "last.json"
|
|
_write_json(receipt, {"receipt": "fixture-state-receipt", "status": "unchanged"})
|
|
|
|
monkeypatch.setattr(policy, "ROOT", runtime_root, raising=False)
|
|
monkeypatch.setattr(policy, "CONFIG_ROOT", config_root, raising=False)
|
|
monkeypatch.setattr(policy, "REGISTRY", registry)
|
|
monkeypatch.setattr(policy, "BACKUP_ROOT", backup_root, raising=False)
|
|
monkeypatch.setattr(policy, "INBOX", inbox)
|
|
monkeypatch.setattr(policy, "PROJECT_LOCK", project_lock)
|
|
monkeypatch.setattr(policy, "GLOBAL_LOCK", global_lock)
|
|
# The production proof shells out to Docker. This suite exercises only the policy
|
|
# reconciliation against temporary fixtures and must never contact a live controller.
|
|
monkeypatch.setattr(policy, "_verify_volume_contract", lambda: "a" * 64)
|
|
return {
|
|
"root": runtime_root,
|
|
"state": state_root,
|
|
"inbox": inbox,
|
|
"registry": registry,
|
|
"project_lock": project_lock,
|
|
"global_lock": global_lock,
|
|
"receipt": receipt,
|
|
"backup_root": backup_root,
|
|
"original": original,
|
|
}
|
|
|
|
|
|
def _tree_bytes(root: Path) -> dict[str, bytes]:
|
|
return {
|
|
path.relative_to(root).as_posix(): path.read_bytes()
|
|
for path in sorted(root.rglob("*"))
|
|
if path.is_file() and not path.is_symlink()
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def root_owned_tmp_path():
|
|
if os.name != "posix" or not hasattr(os, "geteuid"):
|
|
pytest.skip("filesystem contract uses POSIX ownership, flock and fsync")
|
|
if os.geteuid() != 0:
|
|
pytest.skip("filesystem contract requires root-owned temporary fixtures")
|
|
base = Path(tempfile.mkdtemp(prefix="mediarr-validation-policy-", dir="/root"))
|
|
try:
|
|
yield base
|
|
finally:
|
|
shutil.rmtree(base)
|
|
|
|
|
|
def test_accepts_exact_legacy_and_required_policies_without_downgrade() -> None:
|
|
required = policy._validate(_projects(policy.REQUIRED_POLICY))
|
|
legacy = policy._validate(_projects(policy.LEGACY_POLICY))
|
|
assert required == "required"
|
|
assert legacy == "legacy"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutate",
|
|
[
|
|
lambda project: project.update(branch="main"),
|
|
lambda project: project.update(source_dir="/tmp/mediarr"),
|
|
lambda project: project.update(compose_files=["other.yml"]),
|
|
lambda project: project.update(validation={"contexts": ["*"]}),
|
|
lambda project: project.update(
|
|
validation={**policy.REQUIRED_POLICY, "contexts": ["CI / gates (pull_request)"]}
|
|
),
|
|
lambda project: project.update(
|
|
validation={**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + ["extra"]}
|
|
),
|
|
lambda project: project.update(
|
|
validation={**policy.REQUIRED_POLICY, "timeout_seconds": 1800}
|
|
),
|
|
],
|
|
ids=["branch", "source", "compose", "wildcard", "pull-request", "extra-context", "wrong-timeout"],
|
|
)
|
|
def test_rejects_unreviewed_target_or_validation_policy(mutate) -> None:
|
|
projects = _projects(policy.REQUIRED_POLICY)
|
|
mutate(projects[policy.REPOSITORY])
|
|
with pytest.raises(RuntimeError):
|
|
policy._validate(projects)
|
|
|
|
|
|
def test_registry_inline_validator_accepts_only_mediarr_legacy_or_dual_push_policy() -> None:
|
|
compatible = _extract_registry_compatibility()
|
|
legacy = copy.deepcopy(policy.LEGACY_POLICY)
|
|
required = copy.deepcopy(policy.REQUIRED_POLICY)
|
|
other_managed_only = {**legacy, "timeout_seconds": 1800}
|
|
|
|
assert compatible(policy.REPOSITORY, legacy) is True
|
|
assert compatible(policy.REPOSITORY, required) is True
|
|
assert compatible("jens/another-project", other_managed_only) is True
|
|
assert compatible("jens/another-project", required) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"validation",
|
|
[
|
|
{**policy.REQUIRED_POLICY, "contexts": ["*"]},
|
|
{**policy.REQUIRED_POLICY, "contexts": ["CI / gates (pull_request)"]},
|
|
{**policy.REQUIRED_POLICY, "contexts": []},
|
|
{**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + ["extra"]},
|
|
{**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + [PUSH_CONTEXTS[0]]},
|
|
{**policy.REQUIRED_POLICY, "required": False},
|
|
None,
|
|
],
|
|
ids=["wildcard", "pull-request", "missing", "extra", "duplicate", "not-required", "missing-policy"],
|
|
)
|
|
def test_registry_inline_validator_rejects_mediarr_policy_drift(validation) -> None:
|
|
compatible = _extract_registry_compatibility()
|
|
assert compatible(policy.REPOSITORY, validation) is False
|
|
|
|
|
|
def test_registry_inline_validator_rejects_non_mediarr_policy_drift() -> None:
|
|
compatible = _extract_registry_compatibility()
|
|
assert compatible("jens/another-project", {**policy.LEGACY_POLICY, "contexts": ["*"]}) is False
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["check", "apply"])
|
|
@pytest.mark.parametrize("status", ACTIVE_STATUSES)
|
|
def test_reconcile_refuses_active_or_unknown_mediarr_delivery_status(
|
|
root_owned_tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
status: str,
|
|
operation: str,
|
|
) -> None:
|
|
fs = _prepare_filesystem(
|
|
root_owned_tmp_path,
|
|
monkeypatch,
|
|
validation=policy.REQUIRED_POLICY,
|
|
inbox_records={
|
|
"mediarr.json": {"repository": policy.REPOSITORY, "status": status},
|
|
"terminal.json": {"repository": policy.REPOSITORY, "status": "deployed"},
|
|
},
|
|
)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
before_registry = registry.read_bytes()
|
|
before_state = _tree_bytes(fs["state"])
|
|
|
|
with pytest.raises(RuntimeError):
|
|
policy.reconcile(operation)
|
|
|
|
assert registry.read_bytes() == before_registry
|
|
assert _tree_bytes(fs["state"]) == before_state
|
|
|
|
|
|
def test_check_accepts_legacy_without_mutation_and_returns_only_reviewed_contexts(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
before = registry.read_bytes()
|
|
before_identity = registry.stat()
|
|
before_state = _tree_bytes(fs["state"])
|
|
|
|
result = policy.reconcile("check")
|
|
|
|
assert result == {"status": "ready", "contexts": PUSH_CONTEXTS}
|
|
assert registry.read_bytes() == before
|
|
assert (registry.stat().st_dev, registry.stat().st_ino) == (
|
|
before_identity.st_dev,
|
|
before_identity.st_ino,
|
|
)
|
|
assert _tree_bytes(fs["state"]) == before_state
|
|
assert not list(registry.parent.glob("*.mediarr-policy-*"))
|
|
|
|
|
|
def test_apply_updates_only_mediarr_and_keeps_private_backup_and_state_history(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch)
|
|
registry = fs["registry"]
|
|
state = fs["state"]
|
|
backup_root = fs["backup_root"]
|
|
assert (
|
|
isinstance(registry, Path)
|
|
and isinstance(state, Path)
|
|
and isinstance(backup_root, Path)
|
|
)
|
|
original = registry.read_bytes()
|
|
original_projects = json.loads(original)
|
|
state_before = _tree_bytes(state)
|
|
real_replace = policy.os.replace
|
|
temporary_metadata: list[tuple[str, int, int]] = []
|
|
|
|
def inspect_temporary(
|
|
source: os.PathLike[str] | str, target: os.PathLike[str] | str
|
|
) -> None:
|
|
source_path = Path(source)
|
|
info = source_path.stat()
|
|
temporary_metadata.append(
|
|
(source_path.name, stat.S_IMODE(info.st_mode), info.st_uid)
|
|
)
|
|
real_replace(source, target)
|
|
|
|
monkeypatch.setattr(policy.os, "replace", inspect_temporary)
|
|
|
|
result = policy.reconcile("apply")
|
|
|
|
assert set(result) == {"status", "contexts", "backup"}
|
|
assert result["status"] == "updated"
|
|
assert result["contexts"] == PUSH_CONTEXTS
|
|
assert isinstance(result["backup"], str)
|
|
assert Path(result["backup"]).name == result["backup"]
|
|
backup = backup_root / result["backup"]
|
|
assert backup.read_bytes() == original
|
|
assert stat.S_IMODE(backup.stat().st_mode) == 0o600
|
|
assert backup.stat().st_uid == 0
|
|
assert len(temporary_metadata) == 1
|
|
assert temporary_metadata[0][1:] == (0o600, 0)
|
|
current_projects = json.loads(registry.read_bytes())
|
|
assert current_projects[policy.REPOSITORY]["validation"] == policy.REQUIRED_POLICY
|
|
original_projects[policy.REPOSITORY]["validation"] = policy.REQUIRED_POLICY
|
|
assert current_projects == original_projects
|
|
assert stat.S_IMODE(registry.stat().st_mode) == 0o600
|
|
assert registry.stat().st_uid == 0
|
|
assert sorted(path.name for path in registry.parent.iterdir()) == [registry.name]
|
|
assert _tree_bytes(state) == {
|
|
**state_before,
|
|
f"registry-backups/{backup.name}": original,
|
|
}
|
|
assert not list(registry.parent.glob(f".{registry.name}.*.tmp"))
|
|
|
|
|
|
def test_apply_is_idempotent_for_required_policy_but_still_checks_active_delivery(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(
|
|
root_owned_tmp_path, monkeypatch, validation=policy.REQUIRED_POLICY
|
|
)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
before = registry.read_bytes()
|
|
|
|
checked = policy.reconcile("check")
|
|
assert checked == {"status": "ready", "contexts": PUSH_CONTEXTS}
|
|
assert registry.read_bytes() == before
|
|
|
|
result = policy.reconcile("apply")
|
|
|
|
assert result == {"status": "ready", "contexts": PUSH_CONTEXTS}
|
|
assert registry.read_bytes() == before
|
|
assert not list(registry.parent.glob("*.mediarr-policy-*"))
|
|
|
|
|
|
def test_non_root_refuses_before_reading_or_mutating_registry(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
registry = tmp_path / "must-not-be-read.json"
|
|
registry.write_text("sentinel", encoding="utf-8")
|
|
monkeypatch.setattr(policy, "REGISTRY", registry)
|
|
monkeypatch.setattr(policy.os, "geteuid", lambda: 1000, raising=False)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
policy.reconcile("check")
|
|
|
|
assert registry.read_text(encoding="utf-8") == "sentinel"
|
|
|
|
|
|
def test_registry_symlink_is_refused_without_touching_its_target(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
outside = registry.with_name("outside.json")
|
|
outside.write_bytes(registry.read_bytes())
|
|
outside.chmod(0o600)
|
|
registry.unlink()
|
|
registry.symlink_to(outside)
|
|
target_before = outside.read_bytes()
|
|
|
|
with pytest.raises(RuntimeError):
|
|
policy.reconcile("apply")
|
|
|
|
assert outside.read_bytes() == target_before
|
|
|
|
|
|
def test_inbox_symlink_is_refused_even_when_registry_is_already_required(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(
|
|
root_owned_tmp_path,
|
|
monkeypatch,
|
|
validation=policy.REQUIRED_POLICY,
|
|
inbox_records={},
|
|
)
|
|
inbox = fs["inbox"]
|
|
assert isinstance(inbox, Path)
|
|
outside = inbox.with_name("outside-inbox")
|
|
outside.mkdir(mode=0o700)
|
|
inbox.rmdir()
|
|
inbox.symlink_to(outside, target_is_directory=True)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
policy.reconcile("check")
|
|
|
|
|
|
def test_inbox_record_symlink_is_refused_without_touching_its_target(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(
|
|
root_owned_tmp_path,
|
|
monkeypatch,
|
|
validation=policy.REQUIRED_POLICY,
|
|
inbox_records={},
|
|
)
|
|
inbox = fs["inbox"]
|
|
assert isinstance(inbox, Path)
|
|
outside = inbox.with_name("outside-active-delivery.json")
|
|
target_before = _write_json(
|
|
outside,
|
|
{"repository": policy.REPOSITORY, "status": "running"},
|
|
)
|
|
(inbox / "active-delivery.json").symlink_to(outside)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
policy.reconcile("check")
|
|
|
|
assert outside.read_bytes() == target_before
|
|
|
|
|
|
@pytest.mark.parametrize("race", ["content", "identity"])
|
|
def test_apply_refuses_registry_cas_race_before_atomic_replace(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch, race: str
|
|
) -> None:
|
|
fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
original_fsync = policy.os.fsync
|
|
original_replace = policy.os.replace
|
|
original_identity = registry.stat()
|
|
external = registry.with_name("concurrent-registry.json")
|
|
injected = False
|
|
|
|
def race_after_durable_temp(fd: int) -> None:
|
|
nonlocal injected
|
|
original_fsync(fd)
|
|
if injected:
|
|
return
|
|
injected = True
|
|
if race == "content":
|
|
registry.write_bytes(b'{"external":"concurrent update"}\n')
|
|
registry.chmod(0o600)
|
|
else:
|
|
external.write_bytes(registry.read_bytes())
|
|
external.chmod(0o600)
|
|
original_replace(external, registry)
|
|
|
|
monkeypatch.setattr(policy.os, "fsync", race_after_durable_temp)
|
|
with pytest.raises((RuntimeError, OSError)):
|
|
policy.reconcile("apply")
|
|
|
|
assert injected
|
|
assert (
|
|
json.loads(registry.read_bytes()).get(policy.REPOSITORY, {}).get("validation")
|
|
!= policy.REQUIRED_POLICY
|
|
)
|
|
if race == "identity":
|
|
assert (registry.stat().st_dev, registry.stat().st_ino) != (
|
|
original_identity.st_dev,
|
|
original_identity.st_ino,
|
|
)
|
|
assert not list(registry.parent.glob(f".{registry.name}.*.tmp"))
|
|
|
|
|
|
def test_apply_replace_failure_preserves_registry_and_cleans_temporary_file(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch)
|
|
registry = fs["registry"]
|
|
assert isinstance(registry, Path)
|
|
original = registry.read_bytes()
|
|
|
|
def reject_replace(_source: os.PathLike[str] | str, _target: os.PathLike[str] | str) -> None:
|
|
raise OSError("synthetic atomic replace refusal")
|
|
|
|
monkeypatch.setattr(policy.os, "replace", reject_replace)
|
|
with pytest.raises((RuntimeError, OSError)):
|
|
policy.reconcile("apply")
|
|
|
|
assert registry.read_bytes() == original
|
|
assert not list(registry.parent.glob(f".{registry.name}.*.tmp"))
|
|
|
|
|
|
def test_project_lock_is_acquired_before_global_lock(
|
|
root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
if not sys.platform.startswith("linux"):
|
|
pytest.skip("lock-order contention fixture observes Linux flock inode ownership")
|
|
import fcntl
|
|
|
|
fs = _prepare_filesystem(
|
|
root_owned_tmp_path, monkeypatch, validation=policy.REQUIRED_POLICY
|
|
)
|
|
project_lock = fs["project_lock"]
|
|
global_lock = fs["global_lock"]
|
|
assert isinstance(project_lock, Path) and isinstance(global_lock, Path)
|
|
project_identity = (project_lock.stat().st_dev, project_lock.stat().st_ino)
|
|
project_attempted = threading.Event()
|
|
thread_errors: list[RuntimeError] = []
|
|
real_flock = fcntl.flock
|
|
|
|
def observed_flock(file_or_fd, operation: int) -> None:
|
|
descriptor = file_or_fd if isinstance(file_or_fd, int) else file_or_fd.fileno()
|
|
info = os.fstat(descriptor)
|
|
if (info.st_dev, info.st_ino) == project_identity and operation & fcntl.LOCK_EX:
|
|
project_attempted.set()
|
|
real_flock(file_or_fd, operation)
|
|
|
|
monkeypatch.setattr(fcntl, "flock", observed_flock)
|
|
|
|
def run_check() -> None:
|
|
try:
|
|
policy.reconcile("check")
|
|
except RuntimeError as exc: # surfaced in the parent assertion below
|
|
thread_errors.append(exc)
|
|
|
|
with project_lock.open("a+") as held_project:
|
|
real_flock(held_project, fcntl.LOCK_EX)
|
|
thread = threading.Thread(target=run_check, daemon=True)
|
|
thread.start()
|
|
try:
|
|
assert project_attempted.wait(timeout=3), "reconcile never attempted the project lock"
|
|
with global_lock.open("a+") as probe_global:
|
|
try:
|
|
real_flock(probe_global, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
pytest.fail("reconcile acquired the global lock before blocking on the project lock")
|
|
else:
|
|
real_flock(probe_global, fcntl.LOCK_UN)
|
|
finally:
|
|
real_flock(held_project, fcntl.LOCK_UN)
|
|
thread.join(timeout=5)
|
|
|
|
assert not thread.is_alive(), "reconcile remained blocked after the project lock was released"
|
|
assert not thread_errors
|