from __future__ import annotations import copy import importlib.util import sys from pathlib import Path from types import SimpleNamespace import pytest ROOT = Path(__file__).resolve().parents[2] REPOSITORY = "jens/unraid-appops-gateway" REVISION = "c" * 40 @pytest.fixture def recovery(monkeypatch): if sys.platform == "win32": pytest.skip("The root controller uses Linux fcntl") monkeypatch.syspath_prepend(str(ROOT / "ops/gitea-deploy-broker")) spec = importlib.util.spec_from_file_location("signed_rollback_test", ROOT / "ops/gitea-deploy-broker/deploy.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) monkeypatch.setattr(module, "require_host_root", lambda: None) project = {"slug": "unraid-appops-gateway"} metadata = {"id": REVISION[:12], "revision": REVISION, "builtAt": "2026-10-08T00:00:00.000Z"} backup = {"schema": "fixture", "sha256": "b" * 64} prior = {"revision": module.APPOPS_LEGACY_OLD_REVISION} snapshot = { "revision": module.APPOPS_LEGACY_OLD_REVISION, "legacy_provenance": True, "receipt": {}, "appops_runtime_model": prior, "appops_data_backup": backup, } payload = { "repository": REPOSITORY, "revision": REVISION, "kind": "recovery", "phase": "rolling_back", "priorSnapshot": snapshot, "priorModelPointer": prior, "candidateModelPointer": {"revision": REVISION}, "candidateImageId": "sha256:" + "e" * 64, "registryFingerprint": "registry-proof", "metadata": metadata, "transactionId": "transaction-fixture", "preMutationRuntimeSha256": "old-runtime", } pending = {"payload": payload, "sha256": "a" * 64} message = "Expected containers are not healthy/running: fixture; rollback error: AppOps rollback withheld after external runtime drift" state = { "repository": REPOSITORY, "status": "failed", "message": message, "live_revision": module.APPOPS_LEGACY_OLD_REVISION, "live_status": "unknown", "live_receipt": {}, "last_attempt": {"revision": REVISION, "status": "failed", "message": message, "metadata": {**metadata, "appopsDataBackup": backup}}, } identity = {"revision": REVISION, "runtimeCasSha256": "4" * 64, "containerId": "5" * 64, "imageId": payload["candidateImageId"]} reader = SimpleNamespace( PROJECTS_PATH=Path("/fixed/projects.json"), _self_upgrade_pending_model_transaction=lambda: pending, _self_upgrade_verify_rollback_model=lambda *_args: True, _self_upgrade_read_fixed_file=lambda *_args, **_kwargs: b"registry", _self_upgrade_secret=lambda: b"fixture-authority", _self_upgrade_mac=lambda *_args: "registry-proof", _self_upgrade_model_runtime_identity=lambda *_args, **_kwargs: identity, _self_upgrade_source_is_exact_legacy=lambda *_args: "verified-source", ) events = [] monkeypatch.setattr(module, "_appops_status_reader", lambda: reader) monkeypatch.setattr(module, "read_appops_state_preimage", lambda _project: (state, b"state", "state-sha")) monkeypatch.setattr(module, "validate_commit", lambda *_args: REVISION) monkeypatch.setattr(module, "deployment_metadata", lambda *_args, **_kwargs: copy.deepcopy(metadata)) monkeypatch.setattr(module, "verify_appops_data_backup", lambda *_args: events.append("backup-verified")) return module, project, pending, state, identity, reader, events def test_preflight_binds_signed_transaction_state_and_runtime_without_mutating(recovery): module, project, pending, _state, identity, _reader, events = recovery result = module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION) assert result[:2] == (pending, {**identity, "rollbackRuntimeKind": "candidate"}) assert len(result[2]) == 64 assert result[3] == "state-sha" assert events == ["backup-verified"] assert project == {"slug": "unraid-appops-gateway"} @pytest.mark.parametrize("drift", ["repository", "phase", "revision", "snapshot", "registry", "state", "metadata", "identity", "signature", "root"]) def test_recovery_rejects_unproven_authority_before_restore(recovery, monkeypatch, drift): module, project, pending, state, identity, reader, _events = recovery repository = REPOSITORY if drift == "repository": repository = "jens/another-project" elif drift == "phase": pending["payload"]["phase"] = "recovery_required" elif drift == "revision": pending["payload"]["revision"] = "a" * 40 elif drift == "snapshot": pending["payload"]["priorSnapshot"]["legacy_provenance"] = False elif drift == "registry": pending["payload"]["registryFingerprint"] = "changed" elif drift == "state": state["live_status"] = "healthy" elif drift == "metadata": state["last_attempt"]["metadata"]["revision"] = "a" * 40 elif drift == "identity": identity["revision"] = "a" * 40 elif drift == "signature": reader._self_upgrade_pending_model_transaction = lambda: None elif drift == "root": monkeypatch.setattr(module, "require_host_root", lambda: (_ for _ in ()).throw(module.DeployError("root required"))) monkeypatch.setattr(module, "restore_compose_snapshot", lambda *_args: pytest.fail("must not restore")) with pytest.raises(module.DeployError): module.resume_appops_model_rollback(project, repository, ROOT, REVISION, "0" * 64) def test_apply_rechecks_preflight_before_restore_and_commits_verified_model(recovery, monkeypatch): module, project, pending, _state, _identity, _reader, events = recovery fingerprint = module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION)[2] monkeypatch.setattr(module, "_appops_require_effective_rollback_current", lambda _p: events.append("candidate-reverified")) sealed_container = {"Id": "5" * 64, "Image": pending["payload"]["candidateImageId"], "Mounts": []} monkeypatch.setattr(module, "inspect_container", lambda _name: sealed_container) monkeypatch.setattr(module, "_appops_runtime_cas_digest", lambda _container: "4" * 64) def restore(p, *_args): events.append("restored") p["_appops_restored_model_binding"] = {"liveRecordSha256": "verified-live"} p["_appops_legacy_restored"] = True def write(*_args, **kwargs): assert kwargs["expected_state_sha256"] == "state-sha" assert kwargs["live_legacy_provenance"] is True assert project["_appops_runtime_model_binding"]["liveRecordSha256"] == "verified-live" events.append("state-committed") monkeypatch.setattr(module, "restore_compose_snapshot", restore) monkeypatch.setattr(module, "write_state", write) def update(p, phase, bindings): if phase == "rolling_back": assert bindings["rollbackResumeIntent"]["stateSha256"] == "state-sha" assert bindings["rollbackResumeIntent"]["candidateRuntimeSha256"] == "4" * 64 events.append("intent-fsynced") else: events.append("transaction-completed") monkeypatch.setattr(module, "_appops_update_model_transaction", update) monkeypatch.setattr(module, "write_appops_progress", lambda *_args: events.append("progress-completed")) with pytest.raises(module.DeployError, match="preflight changed"): module.resume_appops_model_rollback(project, REPOSITORY, ROOT, REVISION, "0" * 64) assert "restored" not in events result = module.resume_appops_model_rollback(project, REPOSITORY, ROOT, REVISION, fingerprint) assert result["status"] == "rolled-back" assert project["_appops_model_transaction"] == pending assert events[-6:] == ["candidate-reverified", "intent-fsynced", "restored", "state-committed", "transaction-completed", "progress-completed"] @pytest.mark.parametrize("drift", ["state", "runtime", "registry", "transaction"]) def test_apply_does_not_adopt_drift_after_the_approved_preflight(recovery, monkeypatch, drift): module, project, pending, state, identity, reader, _events = recovery fingerprint = module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION)[2] original = module.inspect_appops_model_rollback_recovery calls = 0 def inspect(*args): nonlocal calls result = original(*args) calls += 1 if calls == 1: if drift == "state": monkeypatch.setattr(module, "read_appops_state_preimage", lambda _p: (state, b"different", "different-sha")) elif drift == "runtime": identity["containerId"] = "6" * 64 elif drift == "registry": reader._self_upgrade_mac = lambda *_args: "changed-registry" else: pending["sha256"] = "7" * 64 return result monkeypatch.setattr(module, "inspect_appops_model_rollback_recovery", inspect) monkeypatch.setattr(module, "restore_compose_snapshot", lambda *_args: pytest.fail("unapproved restore")) monkeypatch.setattr(module, "write_state", lambda *_args, **_kwargs: pytest.fail("unapproved state overwrite")) with pytest.raises(module.DeployError): module.resume_appops_model_rollback(project, REPOSITORY, ROOT, REVISION, fingerprint) @pytest.mark.parametrize("kind", ["candidate", "restored", "absent"]) def test_signed_intent_allows_only_its_exact_candidate_or_prior_model_resume(recovery, monkeypatch, kind): module, project, pending, _state, identity, reader, events = recovery payload = pending["payload"] payload["rollbackResumeIntent"] = { "schema": "appops-fixed-555-rollback-resume.v1", "stateSha256": "state-sha", "candidateRuntimeSha256": identity["runtimeCasSha256"], "snapshotSha256": module._canonical_json_sha256(payload["priorSnapshot"]), } def identify(pointer, *_args, **_kwargs): if pointer == payload["candidateModelPointer"]: return identity if kind == "candidate" else None return {**identity, "revision": module.APPOPS_LEGACY_OLD_REVISION} if kind == "restored" else None reader._self_upgrade_model_runtime_identity = identify monkeypatch.setattr(module, "inspect_container", lambda _name: None) monkeypatch.setattr(module, "compose_project_container_names", lambda _name: set()) fingerprint = module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION)[2] def finish(p, *_args): events.append("attest" if kind == "restored" else "restore") p["_appops_restored_model_binding"] = {"revision": module.APPOPS_LEGACY_OLD_REVISION} p["_appops_legacy_restored"] = True monkeypatch.setattr(module, "_appops_attest_restored_legacy_model", finish) monkeypatch.setattr(module, "restore_compose_snapshot", finish) monkeypatch.setattr(module, "write_state", lambda *_args, **_kwargs: events.append("commit")) monkeypatch.setattr(module, "_appops_update_model_transaction", lambda _p, phase, _b: ( events.append("complete") if phase == "rolled_back" else pytest.fail("must not replace immutable intent") )) monkeypatch.setattr(module, "write_appops_progress", lambda *_args: None) module.resume_appops_model_rollback(project, REPOSITORY, ROOT, REVISION, fingerprint) assert events[-3:] == ["attest" if kind == "restored" else "restore", "commit", "complete"] @pytest.mark.parametrize("drift", ["state", "snapshot", "candidate", "foreign-container", "unknown-runtime", "no-intent"]) def test_resume_intent_never_grants_unrelated_runtime_or_state(recovery, monkeypatch, drift): module, project, pending, _state, identity, reader, _events = recovery payload = pending["payload"] intent = {"stateSha256": "state-sha", "candidateRuntimeSha256": identity["runtimeCasSha256"], "snapshotSha256": module._canonical_json_sha256(payload["priorSnapshot"])} payload["rollbackResumeIntent"] = intent if drift == "state": intent["stateSha256"] = "changed" elif drift == "snapshot": intent["snapshotSha256"] = "changed" elif drift == "candidate": identity["runtimeCasSha256"] = "8" * 64 else: reader._self_upgrade_model_runtime_identity = lambda *_args, **_kwargs: None monkeypatch.setattr(module, "inspect_container", lambda _name: {"Id": "foreign"} if drift == "unknown-runtime" else None) monkeypatch.setattr(module, "compose_project_container_names", lambda _name: {"foreign"} if drift == "foreign-container" else set()) if drift == "no-intent": payload.pop("rollbackResumeIntent") with pytest.raises(module.DeployError): module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION) def test_appops_source_restore_removes_only_obsolete_empty_tracked_parents(recovery, tmp_path, monkeypatch): module = recovery[0] target = tmp_path / "source" metadata = target / ".gitea-deploy" metadata.mkdir(parents=True) old = ["scripts/vendor/yaml/new.py", "docs/reviews/new.md", "docs/retained/new.md"] (metadata / "tracked-files.txt").write_text("\n".join(old) + "\n") for name in old: path = target / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text("candidate bytes") (target / "docs/retained/private.txt").write_text("keep untracked bytes") (target / "foreign-empty").mkdir() monkeypatch.setattr(module, "tracked_files", lambda *_args: ["scripts/old.py", "docs/old.md"]) monkeypatch.setattr(module, "extract_revision", lambda *_args: None) module.sync_source(ROOT, module.APPOPS_LEGACY_OLD_REVISION, target, "unraid-appops-gateway") assert not (target / "scripts/vendor").exists() assert not (target / "docs/reviews").exists() assert (target / "docs/retained/private.txt").read_text() == "keep untracked bytes" assert (target / "foreign-empty").is_dir() @pytest.mark.parametrize("drift", [None, "id", "image", "environment", "mounts", "host-config", "network", "inventory", "unsigned", "running", "state", "ports"]) def test_signed_resume_owns_exact_stopped_candidate_only(recovery, monkeypatch, drift): module, project, pending, _state, _identity, reader, _events = recovery current = {"Id": "5" * 64, "Image": pending["payload"]["candidateImageId"], "Config": {"Env": ["SYNTHETIC=unchanged"]}, "HostConfig": {"NetworkMode": "fixed"}, "Mounts": [{"Destination": "/data", "Source": "fixed"}], "State": {"Running": False, "Restarting": False, "Status": "exited"}, "NetworkSettings": {"Ports": {}, "Networks": {}}} payload = pending["payload"] payload["rollbackResumeIntent"] = { "stateSha256": "state-sha", "candidateRuntimeSha256": "4" * 64, "candidateStableRuntimeSha256": module._appops_rollback_stable_runtime_sha256(current), "snapshotSha256": module._canonical_json_sha256(payload["priorSnapshot"]), } reader._self_upgrade_model_runtime_identity = lambda *_args, **_kwargs: None monkeypatch.setattr(module, "inspect_container", lambda _name: current) monkeypatch.setattr(module, "compose_project_container_names", lambda _name: {"Unraid-AppOps-Gateway", "foreign"} if drift == "inventory" else {"Unraid-AppOps-Gateway"}) if drift == "id": current["Id"] = "6" * 64 elif drift == "image": current["Image"] = "sha256:" + "6" * 64 elif drift == "environment": current["Config"]["Env"].append("SYNTHETIC=changed") elif drift == "mounts": current["Mounts"][0]["Source"] = "changed" elif drift == "host-config": current["HostConfig"]["NetworkMode"] = "host" elif drift == "network": current["NetworkSettings"]["Networks"]["foreign"] = {} elif drift == "unsigned": payload.pop("rollbackResumeIntent") elif drift == "running": current["State"]["Running"] = True elif drift == "state": current["State"]["Status"] = "dead" elif drift == "ports": current["NetworkSettings"]["Ports"] = {"1216/tcp": [{"HostIp": "0.0.0.0", "HostPort": "1216"}]} if drift is not None: with pytest.raises(module.DeployError): module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION) else: identity = module.inspect_appops_model_rollback_recovery(project, REPOSITORY, ROOT, REVISION)[1] assert identity["rollbackRuntimeKind"] == "stopped-candidate" assert identity["containerId"] == current["Id"]