from __future__ import annotations import ast import copy import hashlib import importlib.util import json import os import stat from pathlib import Path from types import SimpleNamespace import pytest from test_runner_acceptance_quiescence import module, proven, smart_database # noqa: F401 ROOT = Path(__file__).resolve().parents[2] @pytest.fixture def helper(): spec = importlib.util.spec_from_file_location( "root_recovery_review_test", ROOT / "scripts/recover_ci_acceptance_once.py") helper = importlib.util.module_from_spec(spec) spec.loader.exec_module(helper) return helper @pytest.fixture def partial(helper, module, smart_database, monkeypatch, tmp_path): driver, holder, pg, clients, waits = smart_database baseline = driver.snapshot() registrations = copy.deepcopy(driver.runners[:2]) driver.disable_restart(pg["id"]) driver.term(pg["id"]) for row in driver.runners[:2]: row["disabled"] = True holder["value"] = { "schema": module.SCHEMA, "revision": helper.ORIGIN, "manifestDigest": module._digest(module.OBSERVATION), "phase": "quiescing", "baseline": baseline, "registrations": registrations, "registrationProgress": {"9": "disabled", "22": "disabled"}, "progress": {pg["id"]: "term_intent"}, "guards": driver.capture_guards(), "guardProgress": {"runner": "installed", "host": "installed"}, } class Adapter: def __init__(self, projects): pass def __getattr__(self, name): return getattr(driver, name) def maintenance(self): return driver.maintenance() def snapshot(self): return driver.snapshot() monkeypatch.setattr(module, "ControllerDriver", Adapter) monkeypatch.setattr(helper, "_load", lambda: (module, {})) monkeypatch.setattr(helper, "_installed_proof", lambda: None) monkeypatch.setattr(helper, "_read", lambda *args, **kwargs: b"private reviewed helper") bundle = tmp_path / "bundle" bundle.mkdir(mode=0o700) monkeypatch.setattr(helper, "BUNDLE", bundle) monkeypatch.setattr(helper, "__file__", str(bundle / "recover_ci_acceptance_once.py")) monkeypatch.setattr(helper, "sys", SimpleNamespace(flags=SimpleNamespace(isolated=1), dont_write_bytecode=True)) monkeypatch.setattr(helper, "os", SimpleNamespace(name="posix", geteuid=lambda: 0)) monkeypatch.setattr(helper, "stat", SimpleNamespace( S_IMODE=lambda mode: 0o700)) claims = [] monkeypatch.setattr(helper, "_claim", lambda digest: claims.append(digest)) return driver, holder, pg, clients, claims def test_default_preflight_is_readonly_and_preserves_receipt(helper, partial): driver, holder, _, _, claims = partial before = copy.deepcopy(holder["value"]) effects = list(driver.effects) result = helper.run() assert result["status"] == "partial" and result["effectsPerformed"] is False assert result["candidateRevision"] == helper.CANDIDATE assert result["receiptOrigin"] == helper.ORIGIN and result["normalDeployment"] is False assert not claims and driver.effects == effects and holder["value"] == before def test_reviewed_execution_resumes_same_receipt_and_two_clients(helper, module, partial): driver, holder, pg, clients, claims = partial before = holder["value"] identity = (before["revision"], before["manifestDigest"], module._digest(before["baseline"])) result = helper.run(execute=True, confirmation=helper.CONFIRMATION) assert result["status"] == "quiesced" and result["invocationClaimed"] is True assert result["retryAllowed"] is False and result["normalDeployment"] is False assert len(claims) == 1 assert all(not row["state"]["Running"] for row in [pg, *clients]) after = holder["value"] assert identity == (after["revision"], after["manifestDigest"], module._digest(after["baseline"])) assert after["phase"] == "complete" assert all(effect[0] in {"TERM", "restart", "registration"} for effect in driver.effects) @pytest.mark.parametrize("change", ["missing", "origin", "manifest", "complete", "guarding", "restoring", "guards", "registrations", "cid", "stage"]) def test_foreign_or_new_maintenance_never_executes(helper, partial, change): driver, holder, _, _, claims = partial if change == "missing": holder["value"] = None else: receipt = holder["value"] if change == "origin": receipt["revision"] = "a" * 40 elif change == "manifest": receipt["manifestDigest"] = "a" * 64 elif change in {"complete", "guarding", "restoring"}: receipt["phase"] = change elif change == "guards": receipt["guardProgress"]["host"] = "restored" elif change == "registrations": receipt["registrationProgress"]["22"] = "restored" elif change == "cid": receipt["progress"]["a" * 64] = "term_intent" else: receipt["progress"][next(iter(receipt["progress"]))] = "foreign" effects = list(driver.effects) result = helper.run(execute=True, confirmation=helper.CONFIRMATION) assert result["status"] == "blocked" and result["effectsPerformed"] is False assert not claims and driver.effects == effects @pytest.mark.parametrize("fault", ["queued", "changed-container", "busy-lock", "busy-fleet-lock", "claimed-marker"]) def test_admission_races_before_claim_refuse(helper, partial, monkeypatch, fault): from contextlib import contextmanager driver, _, _, _, claims = partial if fault == "queued": driver.pending_jobs = lambda: [{"status": "queued"}] elif fault == "changed-container": driver.state["rows"][0]["imageId"] = "sha256:" + "0" * 64 elif fault == "busy-lock": @contextmanager def busy(): raise RuntimeError("sensitive unexpected detail") yield driver.maintenance = busy elif fault == "busy-fleet-lock": @contextmanager def busy_fleet(**kwargs): raise RuntimeError("fleet lock busy") yield module = driver.module monkeypatch.setattr(module.fleet, "_migration_lock", busy_fleet) else: def claimed(digest): raise helper.RecoveryRejected("invocation_already_claimed_inspect_only") monkeypatch.setattr(helper, "_claim", claimed) effects = list(driver.effects) result = helper.run(execute=True, confirmation=helper.CONFIRMATION) assert result["status"] == "blocked" and not claims and driver.effects == effects assert "sensitive" not in json.dumps(result) def test_failure_after_claim_is_unknown_never_blind_retry(helper, partial, monkeypatch): driver, _, _, _, claims = partial driver.term = lambda cid: (_ for _ in ()).throw(RuntimeError("token-secret-value")) result = helper.run(execute=True, confirmation=helper.CONFIRMATION) assert len(claims) == 1 and result["status"] == "unknown" assert result["effectsPerformed"] is None and result["retryAllowed"] is False assert "token-secret-value" not in json.dumps(result) @pytest.mark.parametrize("execute,confirmation", [(True, None), (True, "other"), (False, "other")]) def test_confirmation_cannot_be_selected(helper, partial, execute, confirmation): with pytest.raises(helper.RecoveryRejected, match="explicit_fixed_confirmation_required"): helper.run(execute=execute, confirmation=confirmation) assert not partial[-1] def test_fixed_artifact_inventory_and_candidate_hash_match_immutable_code(helper): source = ROOT / "ops/gitea-deploy-broker" tree = ast.parse((source / "status_server.py").read_text()) files = next(ast.literal_eval(node.value) for node in tree.body if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "_CONTROLLER_ARTIFACT_FILES" for target in node.targets)) assert helper.FILES == files # The private bundle must contain Git's LF bytes, not a Windows CRLF checkout. canonical = (source / helper.MODULE_NAME).read_bytes().replace(b"\r\n", b"\n") assert hashlib.sha256(canonical).hexdigest() == helper.CANDIDATE_SHA assert helper.ORIGIN != helper.CANDIDATE @pytest.mark.parametrize("change", ["module", "artifact", "revision", "tree", "version"]) def test_installed_identity_pin_rejects_drift(helper, monkeypatch, change): raws = {name: name.encode() for name in helper.FILES} original_hash = hashlib.sha256(raws[helper.MODULE_NAME]).hexdigest() digest = hashlib.sha256() for name, raw in raws.items(): digest.update(name.encode() + b"\0" + raw + b"\0") monkeypatch.setattr(helper, "ORIGINAL_MODULE_SHA", original_hash) monkeypatch.setattr(helper, "ARTIFACT_SHA", digest.hexdigest()) release = {"apiVersion": "itworx.controller-artifact/v1", "controllerVersion": "2.1.21", "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": helper.ORIGIN, "sourceTree": helper.ORIGIN_TREE, "artifactSha256": helper.ARTIFACT_SHA, "builtAt": "2026-10-10T00:00:00Z"} if change in {"module", "artifact"}: raws[helper.MODULE_NAME if change == "module" else helper.FILES[0]] += b"altered" else: release[{"revision": "sourceRevision", "tree": "sourceTree", "version": "controllerVersion"}[change]] = "different" monkeypatch.setattr(helper, "_read", lambda path, **kwargs: json.dumps(release).encode() if path.name == "controller-release.json" else raws[path.name]) with pytest.raises(helper.RecoveryRejected): helper._installed_proof() def test_candidate_hash_refuses_before_any_import(helper, monkeypatch): monkeypatch.setattr(helper, "_installed_proof", lambda: None) monkeypatch.setattr(helper, "_read", lambda *args, **kwargs: b"foreign candidate") with pytest.raises(helper.RecoveryRejected, match="candidate_bytes_changed"): helper._load() @pytest.mark.parametrize("field", ["root", "isolated", "bytecode"]) def test_untrusted_interpreter_refuses(helper, partial, monkeypatch, field): if field == "root": helper.os.geteuid = lambda: 1000 elif field == "isolated": helper.sys.flags.isolated = 0 else: helper.sys.dont_write_bytecode = False with pytest.raises(helper.RecoveryRejected, match="isolated_root_python_required"): helper.run() assert not partial[-1] @pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root POSIX file safety") @pytest.mark.parametrize("boundary", ["before-link", "after-link"]) def test_claim_publication_failure_is_atomic_and_never_replayed(helper, monkeypatch, tmp_path, boundary): bundle = tmp_path / "bundle" bundle.mkdir(mode=0o700) monkeypatch.setattr(helper, "BUNDLE", bundle) monkeypatch.setattr(helper, "MARKER", bundle / "invoked.json") monkeypatch.setattr(helper, "_trusted_parents", lambda path: None) calls = 0 def fsync(fd): nonlocal calls calls += 1 if calls == (1 if boundary == "before-link" else 2): raise OSError("simulated actual durability failure") os.fsync(fd) proxy = SimpleNamespace(**{name: getattr(os, name) for name in dir(os)}) proxy.fsync = fsync monkeypatch.setattr(helper, "os", proxy) with pytest.raises(OSError): helper._claim("a" * 64) marker = bundle / "invoked.json" assert marker.exists() is (boundary == "after-link") if marker.exists(): assert json.loads(helper._read(marker, private=True))["candidateRevision"] == helper.CANDIDATE with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"): helper._claim("a" * 64) assert not list(bundle.glob(".acceptance-recovery-invocation-*")) @pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root POSIX file safety") def test_private_marker_atomic_once_and_reader_metadata(helper, monkeypatch, tmp_path): bundle = tmp_path / "bundle" bundle.mkdir(mode=0o700) monkeypatch.setattr(helper, "BUNDLE", bundle) monkeypatch.setattr(helper, "MARKER", bundle / "invoked.json") # pytest's sandbox ancestors are not the deployment bundle; test leaf metadata here. monkeypatch.setattr(helper, "_trusted_parents", lambda path: None) helper._claim("a" * 64) marker = bundle / "invoked.json" assert stat.S_IMODE(marker.stat().st_mode) == 0o600 assert json.loads(helper._read(marker, private=True))["receiptOrigin"] == helper.ORIGIN with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"): helper._claim("a" * 64) with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"): helper._claim("b" * 64) for mode in (0o644, 0o640, 0o664, 0o777): marker.chmod(mode) with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"): helper._read(marker, private=True) marker.chmod(0o600) os.chown(marker, 1000, 0) with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"): helper._read(marker, private=True) os.chown(marker, 0, 0) link = bundle / "extra-link" os.link(marker, link) with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"): helper._read(marker, private=True) link.unlink() proxy = SimpleNamespace(**{name: getattr(os, name) for name in dir(os)}) proxy.fstat = lambda fd: SimpleNamespace(**{ key: getattr(os.fstat(fd), key) + (1 if key == "st_ino" else 0) for key in ("st_dev", "st_ino", "st_uid", "st_gid", "st_mode", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns")}) with monkeypatch.context() as scope: scope.setattr(helper, "os", proxy) with pytest.raises(helper.RecoveryRejected, match="recovery_file_changed"): helper._read(marker, private=True) marker.unlink() marker.symlink_to(bundle / "absent") with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"): helper._read(marker, private=True)