from __future__ import annotations import copy import contextlib import hashlib import importlib.util import json import os import subprocess import sys import types from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] BROKER = ROOT / "ops" / "gitea-deploy-broker" APP_OPS_REPOSITORY = "jens/unraid-appops-gateway" RECOVERY_ID = "12345678-1234-4234-8234-123456789abc" LEGACY_555_ENV_KEYS = ( "APPOPS_AUTHENTIK_CLIENT_ID", "APPOPS_AUTHENTIK_CLIENT_SECRET_FILE", "APPOPS_AUTHENTIK_ENTRA_SOURCE", "APPOPS_AUTHENTIK_ISSUER", "APPOPS_AUTH_ADMIN_SUBJECTS", "APPOPS_AUTH_PUBLIC_URL", "APPOPS_BIND_HOST", "APPOPS_DATA_DIR", "APPOPS_DB_PATH", "APPOPS_DEPLOYMENT_MODE", "APPOPS_DISCOVERY_HISTORY_RUNS", "APPOPS_ENABLE_MCP", "APPOPS_ENABLE_N8N", "APPOPS_ENV", "APPOPS_FRONTEND_ORIGIN", "APPOPS_GPU_METRICS_URL", "APPOPS_LOG_LEVEL", "APPOPS_MCP_ALLOWED_HOSTS", "APPOPS_MCP_TOKEN_FILE", "APPOPS_N8N_TOKEN_FILE", "APPOPS_OPERATOR_PROXY_AUTH", "APPOPS_OPERATOR_PROXY_GROUPS", "APPOPS_OPERATOR_PROXY_TRUSTED_NETWORKS", "APPOPS_OPERATOR_PROXY_USERS", "APPOPS_OPERATOR_TOKEN_FILE", "APPOPS_PORT", "APPOPS_PRIVATEMUSE_COMFYUI_MODEL_ROOT", "APPOPS_REQUIRE_OPERATOR_AUTH", "APPOPS_SECRET_KEY_FILE", "APPOPS_STATIC_DIR", "APP_OPS_VOICE_TOKEN", "APP_OPS_VOICE_TOKEN_FILE", "DEPLOYMENT_BROKER_CA_FILE", "DEPLOYMENT_BROKER_URL", "DEPLOYMENT_BROKER_VERIFY_TLS", "GITEA_ADMIN_TOKEN_FILE", "GITEA_RUNNER_TARGETS_JSON", "GITEA_TIMEOUT_SECONDS", "GITEA_URL", "GITEA_VERIFY_TLS", "GPG_KEY", "PATH", "PYTHONDONTWRITEBYTECODE", "PYTHONUNBUFFERED", "PYTHON_SHA256", "PYTHON_VERSION", ) @pytest.fixture def deploy(monkeypatch): monkeypatch.syspath_prepend(str(BROKER)) if "fcntl" not in sys.modules: fcntl = types.ModuleType("fcntl") fcntl.LOCK_EX = 2 fcntl.LOCK_UN = 8 fcntl.LOCK_NB = 4 fcntl.flock = lambda *_args: None monkeypatch.setitem(sys.modules, "fcntl", fcntl) spec = importlib.util.spec_from_file_location( "appops_recovery_deploy_binding_test", BROKER / "deploy.py", ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def _appops_project() -> dict[str, object]: return { "_repository_name": APP_OPS_REPOSITORY, "slug": "unraid-appops-gateway", "gitea_repository_id": 9, "branch": "main", } @pytest.mark.parametrize( ("mode", "repository", "recovery_id", "legacy_id"), [ ("deploy", "jens/other", RECOVERY_ID, None), ("rollback", APP_OPS_REPOSITORY, RECOVERY_ID, None), ("deploy", APP_OPS_REPOSITORY, "not-a-uuid", None), ("deploy", APP_OPS_REPOSITORY, RECOVERY_ID, "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"), ], ) def test_cli_recovery_binding_rejects_before_mutating_git_environment( deploy, monkeypatch, tmp_path, mode, repository, recovery_id, legacy_id, ): project = _appops_project() project["_repository_name"] = repository monkeypatch.setattr(deploy, "load_projects", lambda: {repository: dict(project)}) monkeypatch.setattr(deploy, "repository_path", lambda *_args: tmp_path / "repo.git") monkeypatch.setattr( deploy, "validate_commit", lambda *_args: pytest.fail("invalid recovery binding reached Git validation"), ) monkeypatch.setattr( deploy, "_appops_status_reader", lambda: pytest.fail("invalid recovery binding read the authorization store"), ) monkeypatch.setattr(sys, "argv", ["deploy.py", mode, repository, "c" * 40]) monkeypatch.setenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", recovery_id) if legacy_id is None: monkeypatch.delenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", raising=False) else: monkeypatch.setenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", legacy_id) monkeypatch.setenv("GIT_NO_REPLACE_OBJECTS", "0") with pytest.raises(deploy.DeployError, match="AppOps self-upgrade recovery binding is invalid"): deploy.main() assert os.environ["GIT_NO_REPLACE_OBJECTS"] == "0" def test_valid_cli_recovery_binding_carries_only_authorized_identity(deploy, monkeypatch): project = _appops_project() authorization = {"requestId": RECOVERY_ID, "revision": "c" * 40} reader = types.SimpleNamespace( _self_upgrade_recovery_authorization=lambda request_id, **_kwargs: ( authorization if request_id == RECOVERY_ID else None ) ) monkeypatch.setattr(deploy, "_appops_status_reader", lambda: reader) monkeypatch.setenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", RECOVERY_ID) monkeypatch.delenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", raising=False) assert deploy.bind_appops_self_upgrade_recovery(project, "deploy") == RECOVERY_ID assert project["_appops_self_upgrade_recovery_id"] == RECOVERY_ID assert project["_appops_self_upgrade_recovery_authorization"] == authorization def test_recovery_same_sha_fastpath_waits_for_global_preimage_cas( deploy, monkeypatch, tmp_path, capsys, ): project = { **_appops_project(), "source_dir": str(tmp_path / "prepared-source"), "compose_project": "unraid-appops-gateway", "containers": ["Unraid-AppOps-Gateway"], "compose_files": ["docker-compose.yml"], "verify_oci_revision": True, "environment": {}, } revision = "c" * 40 metadata = { "id": revision[:12], "repository": APP_OPS_REPOSITORY, "revision": revision, "tree": "d" * 40, "treeDigest": "e" * 64, "builtAt": "2026-10-07T12:00:00.000Z", } receipt = {"repository": APP_OPS_REPOSITORY, "revision": revision, "proof": "verified"} state = { "repository": APP_OPS_REPOSITORY, "live_revision": revision, "live_status": "healthy", "live_artifacts": {"appops_runtime_model": {"pointer": "signed"}}, } projects = {APP_OPS_REPOSITORY: project} events = [] global_lock_held = False def acquire_project_lock(*_args, **_kwargs): return contextlib.nullcontext() @contextlib.contextmanager def acquire_global_lock(): nonlocal global_lock_held assert not global_lock_held global_lock_held = True events.append("global-lock-acquired") try: yield finally: global_lock_held = False events.append("global-lock-released") def prepare_recovery(actual_project, repository, _repo, actual_revision, _metadata, _state): assert global_lock_held assert actual_project["_appops_self_upgrade_recovery_id"] == RECOVERY_ID assert repository == APP_OPS_REPOSITORY assert actual_revision == revision events.append("recovery-preimage-cas") def check_current_receipt(repository, _project, actual_revision, tree): assert global_lock_held assert repository == APP_OPS_REPOSITORY assert actual_revision == revision assert tree == metadata["tree"] events.append("same-sha-receipt-check") return receipt monkeypatch.setattr(deploy, "load_projects", lambda: projects) monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, item: dict(item)) monkeypatch.setattr(deploy, "repository_path", lambda *_args: tmp_path / "repo.git") monkeypatch.setattr(deploy, "validate_commit", lambda *_args: revision) monkeypatch.setattr(deploy, "deployment_metadata", lambda *_args: metadata) monkeypatch.setattr(deploy, "validate_project_config", lambda *_args: None) monkeypatch.setattr(deploy, "exclusive_lock", acquire_project_lock) monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", acquire_global_lock) monkeypatch.setattr(deploy, "read_state", lambda _project: state) monkeypatch.setattr( deploy, "state_live_artifacts", lambda value: dict(value.get("live_artifacts") or {}), ) monkeypatch.setattr(deploy, "state_live_revision", lambda _state: revision) monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_args: None) monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_args: None) monkeypatch.setattr(deploy, "_appops_prepare_recovery", prepare_recovery) monkeypatch.setattr(deploy, "_appops_legacy_reconcile_pending_rollback", lambda *_a, **_k: False) monkeypatch.setattr(deploy, "reconcile_appops_deployed_intent", lambda *_args: None) monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_args: None) monkeypatch.setattr(deploy, "verified_current_receipt", check_current_receipt) monkeypatch.setattr( deploy, "sync_source", lambda *_args, **_kwargs: pytest.fail("same-SHA recovery fastpath synced source"), ) monkeypatch.setattr( deploy, "copy_environment", lambda *_args, **_kwargs: pytest.fail("same-SHA recovery fastpath copied environment"), ) reader = types.SimpleNamespace( _self_upgrade_recovery_authorization=lambda request_id, **_kwargs: ( { "requestId": request_id, "revision": revision, "journal": {"revision": revision}, } if request_id == RECOVERY_ID else None ) ) monkeypatch.setattr(deploy, "_appops_status_reader", lambda: reader) monkeypatch.setenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", RECOVERY_ID) monkeypatch.delenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", raising=False) monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", APP_OPS_REPOSITORY, revision]) assert deploy.main() == 0 output = capsys.readouterr().out assert '"status": "already-current"' in output assert events == [ "global-lock-acquired", "recovery-preimage-cas", "same-sha-receipt-check", "global-lock-released", ] def test_effective_runtime_model_preserves_captured_environment_and_changes_only_provenance( deploy, monkeypatch, ): class Adoption: @staticmethod def _inspected_environment(container): parsed = {} for entry in container["Config"]["Env"]: key, value = entry.split("=", 1) if key in parsed: raise ValueError("duplicate environment key") parsed[key] = value return parsed monkeypatch.setattr(deploy, "_load_registered_legacy_adoption", lambda: Adoption()) provenance = { "APPOPS_SOURCE_REVISION": "1" * 40, "APPOPS_SOURCE_REPOSITORY": APP_OPS_REPOSITORY, "APPOPS_SOURCE_TREE": "2" * 40, "APPOPS_BUILD_ID": "333333333333", "APPOPS_BUILD_DATE": "2026-10-07T11:30:00.000Z", } captured = { **provenance, "OPENRGB_HOST": "198.19.1.2", "WATCH_HOST_DIR": "/srv/example/appdata/watch", "MODEL_OVERRIDE": "operator-value-with-$-and-${literal}", "EXTRA_RUNTIME_FLAG": "preserved", } source_model = { "name": "unraid-appops-gateway", "networks": {"default": {}, "appops_proxy": {"external": True}}, "services": { "appops-gateway": { "image": "gitea.invalid/source-build:appops", "environment": {**captured, "SOURCE_ONLY_DEFAULT": "must-not-drop-runtime"}, "env_file": [".env.example", ".env"], "ports": ["127.0.0.1:1216:1216"], "volumes": ["/data:/data"], "healthcheck": {"test": ["CMD", "python", "-m", "health"]}, }, }, } original_model = copy.deepcopy(source_model) metadata = { "id": "abcdef012345", "repository": APP_OPS_REPOSITORY, "revision": "c" * 40, "tree": "d" * 40, "builtAt": "2026-10-07T12:00:00.000Z", } effective = deploy.appops_effective_runtime_model(source_model, captured, metadata) effective_environment = effective["services"]["appops-gateway"]["environment"] assert set(effective_environment) == set(captured) for key, value in captured.items(): if key not in provenance: assert effective_environment[key] == value assert effective_environment["MODEL_OVERRIDE"] == "operator-value-with-$-and-${literal}" assert {key: effective_environment[key] for key in provenance} == { "APPOPS_SOURCE_REVISION": metadata["revision"], "APPOPS_SOURCE_REPOSITORY": metadata["repository"], "APPOPS_SOURCE_TREE": metadata["tree"], "APPOPS_BUILD_ID": metadata["id"], "APPOPS_BUILD_DATE": metadata["builtAt"], } assert source_model == original_model assert "env_file" not in effective["services"]["appops-gateway"] def test_legacy_555_duplicate_environment_is_normalized_but_raw_runtime_stays_cas_bound( deploy, monkeypatch, ): class Adoption: @staticmethod def _inspected_environment(container): parsed = {} for entry in container["Config"]["Env"]: key, value = entry.split("=", 1) if key in parsed: raise ValueError("duplicate environment key") parsed[key] = value return parsed monkeypatch.setattr(deploy, "_load_registered_legacy_adoption", lambda: Adoption()) values = {key: f"synthetic-{index:02d}" for index, key in enumerate(LEGACY_555_ENV_KEYS)} original_origin = values["APPOPS_FRONTEND_ORIGIN"] raw_environment = [f"{key}={value}" for key, value in values.items()] raw_environment.append(f"APPOPS_FRONTEND_ORIGIN={original_origin}") assert len(raw_environment) == 47 assert len(values) == 46 provenance_keys = { "APPOPS_SOURCE_REVISION", "APPOPS_SOURCE_REPOSITORY", "APPOPS_SOURCE_TREE", "APPOPS_BUILD_ID", "APPOPS_BUILD_DATE", } assert not provenance_keys.intersection(values) legacy_container = { "Image": deploy.APPOPS_LEGACY_OLD_IMAGE_ID, "Config": {"Env": raw_environment}, "HostConfig": {}, "NetworkSettings": {}, "Mounts": [], "State": {"Running": True}, } captured = deploy._appops_captured_environment(legacy_container) assert captured == values assert len(captured) == 46 collapsed_runtime = copy.deepcopy(legacy_container) collapsed_runtime["Config"]["Env"] = [f"{key}={value}" for key, value in captured.items()] assert deploy._appops_runtime_cas_digest(legacy_container) != deploy._appops_runtime_cas_digest( collapsed_runtime, ) source_model = {"services": {"appops-gateway": {"environment": {}}}} metadata = { "id": "abcdef012345", "repository": "jens/unraid-appops-gateway", "revision": "c" * 40, "tree": "d" * 40, "builtAt": "2026-10-07T12:00:00.000Z", } effective = deploy.appops_effective_runtime_model(source_model, captured, metadata) candidate_environment = effective["services"]["appops-gateway"]["environment"] assert len(candidate_environment) == 51 assert {key: candidate_environment[key] for key in values} == values assert set(candidate_environment) - set(values) == provenance_keys for image, entries in ( (deploy.APPOPS_LEGACY_OLD_IMAGE_ID, [*raw_environment[:-1], "APPOPS_FRONTEND_ORIGIN=synthetic-conflict"]), ("sha256:" + "8" * 64, raw_environment), ): container = {**legacy_container, "Image": image, "Config": {"Env": entries}} with pytest.raises(deploy.DeployError, match="duplicate keys"): deploy._appops_captured_environment(container) def _appops_prepare_recovery_case(deploy, monkeypatch, tmp_path, capture_change=None): values = {key: f"synthetic-value-{index:02d}" for index, key in enumerate(LEGACY_555_ENV_KEYS)} raw_environment = [f"{key}={value}" for key, value in values.items()] raw_environment.append(f"APPOPS_FRONTEND_ORIGIN={values['APPOPS_FRONTEND_ORIGIN']}") old_image_id = deploy.APPOPS_LEGACY_OLD_IMAGE_ID old_revision = deploy.APPOPS_LEGACY_OLD_REVISION target_revision = "c" * 40 container_id = "a" * 64 state = { "repository": APP_OPS_REPOSITORY, "live_revision": old_revision, "live_status": "healthy", "live_legacy_provenance": True, "live_receipt": {}, "live_artifacts": {"containers": {"Unraid-AppOps-Gateway": { "container_id": container_id, "image_id": old_image_id, }}}, } state_raw = json.dumps(state, sort_keys=True, separators=(",", ":")).encode() + b"\n" state_sha = hashlib.sha256(state_raw).hexdigest() journal = { "recoveryId": RECOVERY_ID, "revision": target_revision, "stateSha256": state_sha, "sourceFingerprint": "b" * 64, "currentContainerId": container_id, "currentImageId": old_image_id, "runtimeEnvironment": values, } container = { "Id": container_id, "Image": old_image_id, "Config": { "Env": raw_environment, "Labels": {"com.docker.compose.project": "unraid-appops-gateway"}, "Hostname": "unraid-appops-gateway", }, "HostConfig": {"RestartPolicy": {"Name": "unless-stopped"}}, "NetworkSettings": {"Ports": {}, "Networks": {"default": {}, "appops_proxy": {}}}, "Mounts": [{ "Type": "bind", "Source": "/srv/example/appdata/unraid-appops-gateway", "Destination": "/data", "RW": True, }], "State": {"Running": True, "Health": {"Status": "healthy", "FailingStreak": 0}}, } image = {"Id": old_image_id, "Config": {"Labels": {"synthetic": "fixed-555"}}} pointer = { "schema": "itworx.appops-effective-compose-model.v1", "recoveryId": RECOVERY_ID, "revision": old_revision, "modelSha256": "1" * 64, "modelRecordSha256": "2" * 64, } source_model = { "name": "unraid-appops-gateway", "services": {"appops-gateway": { "image": deploy.APPOPS_SOURCE_COMPOSE_IMAGE_REF, "environment": {}, }}, } target_metadata = { "id": target_revision[:12], "repository": APP_OPS_REPOSITORY, "revision": target_revision, "tree": "d" * 40, "treeDigest": "e" * 64, "builtAt": "2026-10-07T12:00:00.000Z", } backup = {"schema": "synthetic-backup-proof", "sha256": "f" * 64} captured_snapshot = { "revision": old_revision, "legacy_provenance": True, "receipt": {}, "containers": [{ "name": "Unraid-AppOps-Gateway", "container_id": container_id, "image_id": old_image_id, "image_ref": deploy.APPOPS_LEGACY_OLD_IMAGE_REF, "rollback_ref": f"gitea-deploy-rollback/unraid-appops-gateway-0:{old_revision[:12]}", "was_running": True, }], } if capture_change is not None: capture_change(captured_snapshot, container_id, old_image_id) calls = {"authorizations": [], "source_checks": [], "capture": [], "models": [], "backup": [], "begin": [], "tags": []} class Adoption: @staticmethod def _inspected_environment(observed): result = {} for entry in observed["Config"]["Env"]: key, value = entry.split("=", 1) if key in result: raise ValueError("test expects normalized unique environment") result[key] = value return result class StatusReader: @staticmethod def _self_upgrade_recovery_authorization(request_id, **_kwargs): calls["authorizations"].append(request_id) if request_id != RECOVERY_ID: return None return {"requestId": RECOVERY_ID, "revision": target_revision, "journal": journal} @staticmethod def _self_upgrade_store_compose_model(recovery_id, model_revision, model): calls["models"].append((recovery_id, model_revision, copy.deepcopy(model))) return pointer @staticmethod def _self_upgrade_load_compose_model(recovery_id, model_revision): if recovery_id != RECOVERY_ID or model_revision != old_revision or not calls["models"]: return None return {"pointer": pointer, "model": calls["models"][-1][2]} project = { "_repository_name": APP_OPS_REPOSITORY, "_appops_self_upgrade_recovery_id": RECOVERY_ID, "slug": "unraid-appops-gateway", "source_dir": str(tmp_path / "source"), "compose_project": "unraid-appops-gateway", "containers": ["Unraid-AppOps-Gateway"], } monkeypatch.setattr(deploy, "_load_registered_legacy_adoption", lambda: Adoption()) monkeypatch.setattr(deploy, "_appops_status_reader", lambda: StatusReader()) monkeypatch.setattr( deploy, "verify_appops_legacy_source", lambda actual_repo, actual_project, revision: calls["source_checks"].append( (actual_repo, actual_project, revision) ), ) monkeypatch.setattr( deploy, "read_appops_state_preimage", lambda _project: (copy.deepcopy(state), state_raw, state_sha), ) monkeypatch.setattr(deploy, "inspect_container", lambda _name: copy.deepcopy(container)) monkeypatch.setattr(deploy, "inspect_image", lambda _image_id: copy.deepcopy(image)) monkeypatch.setattr(deploy, "verify_appops_legacy_runtime", lambda *_args, **_kwargs: None) monkeypatch.setattr(deploy, "appops_legacy_compose_model", lambda *_args: (source_model, "0" * 64, "1" * 64)) monkeypatch.setattr( deploy, "sync_source", lambda *_args, **_kwargs: pytest.fail("recovery capture must not sync source"), ) monkeypatch.setattr( deploy, "copy_environment", lambda *_args, **_kwargs: pytest.fail("recovery capture must not copy source environment"), ) def capture_rollback(actual_project, *, previous_state_override, tag_snapshot_images): calls["capture"].append((actual_project, copy.deepcopy(previous_state_override), tag_snapshot_images)) return copy.deepcopy(captured_snapshot) monkeypatch.setattr(deploy, "capture_compose_rollback", capture_rollback) def capture_backup(candidate_revision, previous_revision): calls["backup"].append((candidate_revision, previous_revision)) return backup monkeypatch.setattr(deploy, "capture_appops_data_backup", capture_backup) def begin_transaction(actual_project, revision, metadata, source_fingerprint, kind): calls["begin"].append(( actual_project["_repository_name"], revision, copy.deepcopy(metadata), source_fingerprint, kind, )) assert actual_project["_compose_rollback"]["containers"] == [{ **captured_snapshot["containers"][0], "restore_image_ref": deploy.APPOPS_SOURCE_COMPOSE_IMAGE_REF, }] assert actual_project["_compose_rollback"]["appops_runtime_model"] == pointer assert actual_project["_appops_pre_mutation_runtime_sha256"] == deploy._appops_runtime_cas_digest(container) assert actual_project["_appops_self_upgrade_recovery_id"] == RECOVERY_ID assert actual_project["_appops_self_upgrade_recovery_authorization"]["journal"] is journal stored_environment = calls["models"][-1][2]["services"]["appops-gateway"]["environment"] assert stored_environment == values assert len(stored_environment) == 46 assert not { "APPOPS_SOURCE_REVISION", "APPOPS_SOURCE_REPOSITORY", "APPOPS_SOURCE_TREE", "APPOPS_BUILD_ID", "APPOPS_BUILD_DATE", }.intersection(stored_environment) monkeypatch.setattr(deploy, "_appops_begin_model_transaction", begin_transaction) monkeypatch.setattr(deploy, "run", lambda *args, **kwargs: calls["tags"].append((args, kwargs))) return types.SimpleNamespace( project=project, repo=tmp_path / "repository.git", revision=target_revision, metadata=target_metadata, state=state, state_sha=state_sha, container=container, image=image, raw_environment=raw_environment, values=values, captured_snapshot=captured_snapshot, calls=calls, ) def test_prepare_recovery_persists_exact_555_environment_and_adds_restore_ref_before_transaction( deploy, monkeypatch, tmp_path, ): case = _appops_prepare_recovery_case(deploy, monkeypatch, tmp_path) assert len(case.raw_environment) == 47 assert len(case.values) == 46 deploy._appops_prepare_recovery( case.project, APP_OPS_REPOSITORY, case.repo, case.revision, case.metadata, case.state, ) snapshot_item = case.project["_compose_rollback"]["containers"][0] assert "restore_image_ref" not in case.captured_snapshot["containers"][0] assert snapshot_item["restore_image_ref"] == deploy.APPOPS_SOURCE_COMPOSE_IMAGE_REF assert case.project["_compose_rollback"]["appops_runtime_model"]["recoveryId"] == RECOVERY_ID assert case.project["_compose_rollback"]["appops_runtime_model"]["revision"] == deploy.APPOPS_LEGACY_OLD_REVISION assert case.project["_appops_pre_mutation_runtime_sha256"] == deploy._appops_runtime_cas_digest(case.container) collapsed_container = copy.deepcopy(case.container) collapsed_container["Config"]["Env"] = [ f"{key}={value}" for key, value in case.values.items() ] assert deploy._appops_runtime_cas_digest(case.container) != deploy._appops_runtime_cas_digest( collapsed_container, ) assert case.calls["capture"] == [(case.project, case.state, False)] assert case.calls["begin"] == [( APP_OPS_REPOSITORY, case.revision, case.metadata, "b" * 64, "recovery", )] assert case.calls["authorizations"] == [RECOVERY_ID, RECOVERY_ID] assert case.calls["backup"] == [(case.revision, deploy.APPOPS_LEGACY_OLD_REVISION)] assert case.calls["tags"] == [] @pytest.mark.parametrize("mismatch", ["image", "container_id", "name"]) def test_prepare_recovery_rejects_changed_capture_identity_before_transaction_or_tagging( deploy, monkeypatch, tmp_path, mismatch, ): def alter_snapshot(snapshot, container_id, image_id): item = snapshot["containers"][0] if mismatch == "image": item["image_id"] = "sha256:" + "8" * 64 elif mismatch == "container_id": item["container_id"] = "c" * 64 elif mismatch == "name": item["name"] = "unexpected-container" case = _appops_prepare_recovery_case(deploy, monkeypatch, tmp_path, alter_snapshot) with pytest.raises(deploy.DeployError, match="rollback snapshot changed during capture"): deploy._appops_prepare_recovery( case.project, APP_OPS_REPOSITORY, case.repo, case.revision, case.metadata, case.state, ) assert case.calls["begin"] == [] assert case.calls["tags"] == [] assert case.calls["backup"] == [] assert case.project.get("_compose_rollback") is None def test_capture_compose_rollback_wrapper_forwards_tagging_choice(deploy, monkeypatch): project = {"compose_project": "appops-test", "containers": ["appops-gateway"]} state = {"live_revision": "c" * 40, "live_status": "healthy"} calls = [] def capture_inner(actual_project, **kwargs): calls.append((actual_project, kwargs)) return {"snapshot": "captured"} monkeypatch.setattr(deploy, "_capture_compose_rollback_inner", capture_inner) result = deploy.capture_compose_rollback( project, previous_state_override=state, tag_snapshot_images=False, ) assert result == {"snapshot": "captured"} assert calls == [( project, {"previous_state_override": state, "tag_snapshot_images": False}, )] def _git(path: Path, *args: str) -> str: return subprocess.run( ["git", "-C", str(path), *args], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ).stdout.strip() def _effective_candidate_case(deploy, tmp_path, monkeypatch, drift): checkout = tmp_path / "source" checkout.mkdir() _git(checkout, "init", "-q") _git(checkout, "config", "user.email", "test@example.invalid") _git(checkout, "config", "user.name", "AppOps test") (checkout / "server").mkdir() (checkout / "server/Dockerfile").write_text("FROM scratch\nCOPY . /app\n") (checkout / "server/main.py").write_text("print('tracked candidate')\n") _git(checkout, "add", ".") _git(checkout, "commit", "-qm", "protected candidate") revision = _git(checkout, "rev-parse", "HEAD") bare = tmp_path / "repository.git" subprocess.run( ["git", "clone", "--bare", "--quiet", str(checkout), str(bare)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) (checkout / "server/stale.py").write_text("untracked source must not enter image\n") (checkout / ".env").write_text("PRIVATE_VALUE=excluded\n") source_model = { "services": { "appops-gateway": { "build": {"context": str(checkout), "dockerfile": "server/Dockerfile"}, }, }, } project = { **_appops_project(), "source_dir": str(checkout), "compose_dir": "", "compose_project": "unraid-appops-gateway", "containers": ["Unraid-AppOps-Gateway"], "_repository_path": str(bare), "_registry_project": "internal-test-identity", } exact_root = tmp_path / "controller-builds" monkeypatch.setattr(deploy, "EXACT_SOURCE_ROOT", exact_root) build_context = deploy._load_registered_build_context().prepare_build_contexts( bare, revision, project, source_model, exact_root, )["appops-gateway"] effective_model = copy.deepcopy(source_model) effective_model["services"]["appops-gateway"]["build"]["context"] = build_context pointer = { "recoveryId": RECOVERY_ID, "revision": revision, "modelSha256": "a" * 64, "modelRecordSha256": "b" * 64, } def load_effective_model(recovery_id, value): if drift == "private_model": return None if recovery_id == RECOVERY_ID and value == revision: return {"model": effective_model, "pointer": pointer} return None def fetch_green(_repository, value, _policy): calls["green"] += 1 calls["green_revision"] = value return "pass", {"revision": value, "run": "canonical-full"} reader = types.SimpleNamespace( _validation_policy=lambda _repository, _project: {"policy": "current-full"}, _self_upgrade_protected_head=lambda value: value == revision, _is_current_branch_head=lambda _repository, _project, value: value == revision, _fetch_gitea_validation=fetch_green, _self_upgrade_load_compose_model=load_effective_model, ) registered_project = {"slug": "unraid-appops-gateway", "branch": "main"} project["_registry_project"] = registered_project project["_appops_registry_project"] = copy.deepcopy(registered_project) project["_appops_effective_model_pointer"] = dict(pointer) project["_appops_candidate_source_model"] = copy.deepcopy(source_model) raw_state = b'{"repository":"jens/unraid-appops-gateway","status":"success"}\n' original_state_sha = hashlib.sha256(raw_state).hexdigest() project["_appops_runtime_model_state_sha256"] = original_state_sha runtime = { "Id": "container-before-build", "Image": "sha256:" + "c" * 64, "Config": {"Env": ["MODEL=known", "TOKEN=private"], "Labels": {"x": "y"}}, "HostConfig": {"Memory": 0, "Binds": ["/data:/data"]}, "NetworkSettings": {"Networks": {"default": {"NetworkID": "network-before"}}}, "Mounts": [{"Type": "volume", "Name": "appops-data", "Destination": "/data"}], "State": {"Running": True, "Health": {"Status": "healthy"}}, } project["_appops_pre_mutation_runtime_sha256"] = deploy._appops_runtime_cas_digest(runtime) state_observation = {"raw": raw_state, "sha": original_state_sha} projects = {APP_OPS_REPOSITORY: registered_project} calls = {"commands": [], "green": 0, "green_revision": None} def read_preimage(_project): if drift == "state" and "build" in calls["commands"]: changed = b'{"repository":"jens/unraid-appops-gateway","status":"other"}\n' return {"repository": APP_OPS_REPOSITORY}, changed, hashlib.sha256(changed).hexdigest() return {"repository": APP_OPS_REPOSITORY}, state_observation["raw"], state_observation["sha"] def load_projects(): if drift == "registry" and "build" in calls["commands"]: return {APP_OPS_REPOSITORY: {**registered_project, "branch": "release"}} return projects def mutate_after_build(): if drift == "environment": runtime["Config"]["Env"].append("MODEL=external-change") elif drift == "config": runtime["Config"]["Labels"]["x"] = "externally-changed" elif drift == "hostconfig": runtime["HostConfig"]["Memory"] = 128 elif drift == "container_id": runtime["Id"] = "container-replaced-during-build" elif drift == "image": runtime["Image"] = "sha256:" + "d" * 64 elif drift == "network": runtime["NetworkSettings"]["Networks"]["default"]["NetworkID"] = "network-drifted" elif drift == "mount": runtime["Mounts"][0]["Source"] = "/unexpected/volume" elif drift == "running": runtime["State"]["Running"] = False elif drift == "archive": Path(build_context, "server/main.py").write_text("tampered after build\n") def run(command, **_kwargs): calls["commands"].append(command[-1]) if command[-1] == "build": assert not (Path(build_context) / "server/stale.py").exists() assert not (Path(build_context) / ".env").exists() assert (Path(build_context) / "server/main.py").read_text() == "print('tracked candidate')\n" mutate_after_build() if "up" in command: calls["commands"].append("up-mutated") return None monkeypatch.setattr(deploy, "_appops_status_reader", lambda: reader) monkeypatch.setattr(deploy, "read_appops_state_preimage", read_preimage) monkeypatch.setattr(deploy, "load_projects", load_projects) monkeypatch.setattr(deploy, "inspect_container", lambda _name: runtime) monkeypatch.setattr(deploy, "deployment_environment", lambda *_args, **_kwargs: {}) monkeypatch.setattr(deploy, "compose_command", lambda _project: (checkout, ["docker", "compose"])) monkeypatch.setattr(deploy, "run", run) monkeypatch.setattr(deploy, "prepare_registered_legacy_build", lambda *_args, **_kwargs: None) monkeypatch.setattr(deploy, "write_appops_progress", lambda *_args, **_kwargs: None) monkeypatch.setattr(deploy, "prepare_compose_migration", lambda *_args, **_kwargs: []) monkeypatch.setattr(deploy, "finish_compose_migration", lambda *_args, **_kwargs: None) monkeypatch.setattr(deploy, "workmesh_transition_enabled", lambda _project: False) return project, revision, runtime, calls, original_state_sha, reader @pytest.mark.parametrize( "drift", [ "environment", "config", "hostconfig", "container_id", "image", "network", "mount", "running", "state", "registry", "archive", "private_model", ], ) def test_deploy_project_rejects_after_build_drift_before_up_without_rollback( deploy, monkeypatch, tmp_path, drift, ): project, revision, runtime, calls, _state_sha, _reader = _effective_candidate_case( deploy, tmp_path, monkeypatch, drift, ) with pytest.raises(deploy.AppOpsEffectiveNoMutation): deploy.deploy_project(project, revision, APP_OPS_REPOSITORY) assert calls["commands"] == ["--quiet", "build"] assert calls["green"] == 1 assert calls["green_revision"] == revision if drift in {"environment", "hostconfig", "container_id", "network"}: assert runtime["Id"] in {"container-before-build", "container-replaced-during-build"} @pytest.mark.parametrize("proof_failure", ["protected_head", "branch_head", "full_validation"]) def test_deploy_project_requires_canonical_green_validation_before_up( deploy, monkeypatch, tmp_path, proof_failure, ): project, revision, _runtime, calls, _state_sha, reader = _effective_candidate_case( deploy, tmp_path, monkeypatch, "none", ) if proof_failure == "protected_head": reader._self_upgrade_protected_head = lambda _value: False elif proof_failure == "branch_head": reader._is_current_branch_head = lambda *_args: False else: def fetch_failed(*_args): calls["green"] += 1 calls["green_revision"] = revision return "fail", {"reason": "full validation failed"} reader._fetch_gitea_validation = fetch_failed with pytest.raises(deploy.AppOpsEffectiveNoMutation): deploy.deploy_project(project, revision, APP_OPS_REPOSITORY) assert calls["commands"] == ["--quiet", "build"] assert calls["green"] == (1 if proof_failure == "full_validation" else 0) assert calls["green_revision"] == (revision if proof_failure == "full_validation" else None) def test_actual_registered_archive_is_clean_and_rejects_cached_byte_tampering( deploy, tmp_path, monkeypatch, ): project, revision, _runtime, _calls, _state_sha, _reader = _effective_candidate_case( deploy, tmp_path, monkeypatch, "none", ) model = project["_appops_candidate_source_model"] context = deploy._load_registered_build_context().prepare_build_contexts( Path(project["_repository_path"]), revision, project, model, Path(deploy.EXACT_SOURCE_ROOT), )["appops-gateway"] tree = Path(context) assert (tree / "server/main.py").read_text() == "print('tracked candidate')\n" assert not (tree / "server/stale.py").exists() assert not (tree / ".env").exists() (tree / "server/main.py").write_text("changed archive bytes\n") with pytest.raises(deploy._load_registered_build_context().RegisteredBuildContextError): deploy._load_registered_build_context().prepare_build_contexts( Path(project["_repository_path"]), revision, project, model, Path(deploy.EXACT_SOURCE_ROOT), ) def test_main_does_not_clean_up_foreign_drift_after_effective_pre_mutation_rejection( deploy, monkeypatch, tmp_path, capsys, ): revision = "c" * 40 prior_revision = "d" * 40 raw_state = b'{"repository":"jens/unraid-appops-gateway","status":"success"}\n' state = { "repository": APP_OPS_REPOSITORY, "live_revision": prior_revision, "live_status": "healthy", "live_artifacts": {"appops_runtime_model": {"pointer": "signed-model"}}, } registered = { **_appops_project(), "source_dir": str(tmp_path / "source"), "compose_project": "unraid-appops-gateway", "containers": ["Unraid-AppOps-Gateway"], "compose_files": ["docker-compose.yml"], "verify_oci_revision": True, "environment": {}, } projects = {APP_OPS_REPOSITORY: registered} cleanup_calls = [] progress = [] @contextlib.contextmanager def held_lock(*_args, **_kwargs): yield def no_cleanup(name): def fail_if_called(*_args, **_kwargs): cleanup_calls.append(name) pytest.fail(f"foreign drift rejection entered {name}") return fail_if_called def prepare_followup(actual_project, *_args, **_kwargs): actual_project["_compose_rollback"] = {"containers": [{"name": "Unraid-AppOps-Gateway"}]} actual_project["_appops_runtime_model_state_sha256"] = hashlib.sha256(raw_state).hexdigest() actual_project["_appops_data_backup"] = {"sha256": "e" * 64} return None monkeypatch.setattr(deploy, "load_projects", lambda: projects) monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, item: dict(item)) monkeypatch.setattr(deploy, "repository_path", lambda *_args: tmp_path / "repo.git") monkeypatch.setattr(deploy, "validate_commit", lambda *_args: revision) monkeypatch.setattr(deploy, "deployment_metadata", lambda *_args: { "id": revision[:12], "repository": APP_OPS_REPOSITORY, "revision": revision, "tree": "f" * 40, "treeDigest": "a" * 64, "builtAt": "2026-10-07T12:00:00.000Z", }) monkeypatch.setattr(deploy, "validate_project_config", lambda *_args: None) monkeypatch.setattr(deploy, "exclusive_lock", lambda *_args, **_kwargs: held_lock()) monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: held_lock()) monkeypatch.setattr(deploy, "read_state", lambda _project: state) monkeypatch.setattr(deploy, "state_live_artifacts", lambda value: dict(value.get("live_artifacts") or {})) monkeypatch.setattr(deploy, "state_live_revision", lambda _value: prior_revision) monkeypatch.setattr(deploy, "state_live_status", lambda _value: "healthy") monkeypatch.setattr(deploy, "state_live_legacy_provenance", lambda _value: False) monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_args: None) monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_args: None) monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_args: None) monkeypatch.setattr(deploy, "_appops_prepare_model_followup", prepare_followup) monkeypatch.setattr(deploy, "_appops_prepare_effective_candidate", lambda *_args: None) monkeypatch.setattr(deploy, "_appops_legacy_reconcile_pending_rollback", lambda *_args, **_kwargs: False) monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_args: None) monkeypatch.setattr(deploy, "reconcile_appops_deployed_intent", lambda *_args: None) monkeypatch.setattr(deploy, "revalidate_deployment_candidate", lambda *_args: None) monkeypatch.setattr(deploy, "validate_managed_v1_runtime", lambda *_args: None) monkeypatch.setattr(deploy, "capture_appops_data_backup", lambda *_args: pytest.fail("unexpected backup")) monkeypatch.setattr(deploy, "ensure_docker_capacity", lambda: None) monkeypatch.setattr(deploy, "write_appops_progress", lambda _revision, phase, **_kwargs: progress.append(phase)) monkeypatch.setattr( deploy, "deploy_project", lambda *_args: (_ for _ in ()).throw(deploy.AppOpsEffectiveNoMutation("runtime changed after build")), ) monkeypatch.setattr(deploy, "write_state", no_cleanup("write_state")) monkeypatch.setattr(deploy, "restore_compose_snapshot", no_cleanup("restore_compose_snapshot")) monkeypatch.setattr(deploy, "restore_live_source", no_cleanup("restore_live_source")) monkeypatch.setattr(deploy, "restore_compose_migration", no_cleanup("restore_compose_migration")) monkeypatch.setattr(deploy, "cleanup_failed_first_deploy", no_cleanup("cleanup_failed_first_deploy")) monkeypatch.setattr(deploy, "_appops_legacy_checkpoint_forward_rollback", no_cleanup("legacy_rollback_checkpoint")) monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", APP_OPS_REPOSITORY, revision]) monkeypatch.delenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", raising=False) monkeypatch.delenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", raising=False) with pytest.raises(deploy.AppOpsEffectiveNoMutation, match="runtime changed after build"): deploy.main() assert cleanup_calls == [] assert state["live_revision"] == prior_revision assert progress == ["preflighting"] assert capsys.readouterr().out == "" def test_followup_transaction_journal_failure_precedes_all_source_runtime_and_state_writes( deploy, monkeypatch, tmp_path, ): revision = "c" * 40 prior_revision = "d" * 40 source = tmp_path / "source" source.mkdir() marker = source / "deployed-marker.txt" marker.write_text("prior source remains\n") raw_state = b'{"repository":"jens/unraid-appops-gateway","status":"success"}\n' state = { "repository": APP_OPS_REPOSITORY, "live_revision": prior_revision, "live_status": "healthy", "live_artifacts": {"appops_runtime_model": {"pointer": "signed-model"}}, } registered = { **_appops_project(), "source_dir": str(source), "compose_project": "unraid-appops-gateway", "containers": ["Unraid-AppOps-Gateway"], "compose_files": ["docker-compose.yml"], "verify_oci_revision": True, "environment": {}, } projects = {APP_OPS_REPOSITORY: registered} pointer = { "recoveryId": RECOVERY_ID, "revision": prior_revision, "modelSha256": "a" * 64, "modelRecordSha256": "b" * 64, } effective_model = {"services": {"appops-gateway": {"environment": {"MODEL": "kept"}}}} events = [] runtime = { "Id": "container-stable", "Image": "sha256:" + "c" * 64, "Config": {"Env": ["MODEL=kept"], "Labels": {}}, "HostConfig": {"Binds": ["/data:/data"]}, "NetworkSettings": {"Networks": {"default": {"NetworkID": "network-stable"}}}, "Mounts": [{"Type": "volume", "Name": "data", "Destination": "/data"}], "State": {"Running": True, "Health": {"Status": "healthy"}}, } class StatusReader: @staticmethod def _validation_policy(_repository, _project): return {"policy": "fixed-main-full"} @staticmethod def _self_upgrade_protected_head(value): return value == revision @staticmethod def _is_current_branch_head(_repository, _project, value): return value == revision @staticmethod def _fetch_gitea_validation(_repository, value, _policy): events.append(("green", value)) return "pass", {"revision": value} @staticmethod def _self_upgrade_active_compose_model(_state, _runtime): return { "pointer": pointer, "live": {"sourceFingerprint": "e" * 64}, } @staticmethod def _self_upgrade_load_compose_model(recovery_id, value): if recovery_id == RECOVERY_ID and value == prior_revision: return {"model": effective_model, "pointer": pointer} return None @contextlib.contextmanager def held_lock(*_args, **_kwargs): yield def forbidden(name): def fail(*_args, **_kwargs): events.append(("forbidden", name)) pytest.fail(f"journal failure reached {name}") return fail def begin_transaction(*_args, **_kwargs): events.append(("journal",)) raise deploy.DeployError("test journal store unavailable") monkeypatch.setattr(deploy, "load_projects", lambda: projects) monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, item: dict(item)) monkeypatch.setattr(deploy, "repository_path", lambda *_args: tmp_path / "repo.git") monkeypatch.setattr(deploy, "validate_commit", lambda *_args: revision) monkeypatch.setattr(deploy, "deployment_metadata", lambda *_args: { "id": revision[:12], "repository": APP_OPS_REPOSITORY, "revision": revision, "tree": "f" * 40, "treeDigest": "a" * 64, "builtAt": "2026-10-07T12:00:00.000Z", }) monkeypatch.setattr(deploy, "validate_project_config", lambda *_args: None) monkeypatch.setattr(deploy, "exclusive_lock", lambda *_args, **_kwargs: held_lock()) monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: held_lock()) monkeypatch.setattr(deploy, "read_state", lambda _project: state) monkeypatch.setattr(deploy, "state_live_artifacts", lambda value: dict(value.get("live_artifacts") or {})) monkeypatch.setattr(deploy, "state_live_revision", lambda _value: prior_revision) monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_args: None) monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_args: None) monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_args: None) monkeypatch.setattr(deploy, "_appops_status_reader", StatusReader) monkeypatch.setattr( deploy, "read_appops_state_preimage", lambda _project: (state, raw_state, hashlib.sha256(raw_state).hexdigest()), ) monkeypatch.setattr(deploy, "inspect_container", lambda _name: runtime) monkeypatch.setattr( deploy, "capture_compose_rollback", lambda _project, **_kwargs: (events.append(("snapshot",)) or {"containers": [{"name": "Unraid-AppOps-Gateway"}]}), ) monkeypatch.setattr( deploy, "capture_appops_data_backup", lambda *_args: (events.append(("backup",)) or {"sha256": "1" * 64}), ) monkeypatch.setattr(deploy, "_appops_begin_model_transaction", begin_transaction) monkeypatch.setattr(deploy, "sync_source", forbidden("sync_source")) monkeypatch.setattr(deploy, "copy_environment", forbidden("copy_environment")) monkeypatch.setattr(deploy, "run", forbidden("compose/docker")) monkeypatch.setattr(deploy, "write_state", forbidden("write_state")) monkeypatch.setattr(deploy, "deploy_project", forbidden("deploy_project")) monkeypatch.setattr(deploy, "restore_compose_snapshot", forbidden("restore_compose_snapshot")) monkeypatch.setattr(deploy, "restore_live_source", forbidden("restore_live_source")) monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", APP_OPS_REPOSITORY, revision]) monkeypatch.delenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", raising=False) monkeypatch.delenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", raising=False) with pytest.raises(deploy.DeployError, match="test journal store unavailable"): deploy.main() assert [event[0] for event in events] == ["green", "snapshot", "backup", "journal"] assert marker.read_text() == "prior source remains\n" assert state["live_revision"] == prior_revision def test_prepare_restored_recovery_rechecks_signed_live_state_and_recaptures_under_new_id( deploy, monkeypatch, tmp_path, ): case = _appops_prepare_recovery_case(deploy, monkeypatch, tmp_path) calls = [] def verify(_project, container, image, **kwargs): calls.append(kwargs) assert container["Id"] != deploy.APPOPS_LEGACY_OLD_CONTAINER_ID monkeypatch.setattr(deploy, "verify_appops_legacy_runtime", verify) deploy._appops_prepare_recovery( case.project, APP_OPS_REPOSITORY, case.repo, case.revision, case.metadata, case.state, ) assert calls == [{"require_original_container": False, "recovery_state": case.state}] assert case.calls["models"][0][0] == RECOVERY_ID assert case.calls["models"][0][2]["services"]["appops-gateway"]["environment"] == case.values assert case.project["_compose_rollback"]["containers"][0]["container_id"] == case.container["Id"] def test_prepare_recovery_rejects_environment_drift_before_model_backup_or_transaction( deploy, monkeypatch, tmp_path, ): case = _appops_prepare_recovery_case(deploy, monkeypatch, tmp_path) case.container["Config"]["Env"].append("UNEXPECTED_SETTING=private-sentinel") with pytest.raises(deploy.DeployError, match="complete runtime environment changed"): deploy._appops_prepare_recovery( case.project, APP_OPS_REPOSITORY, case.repo, case.revision, case.metadata, case.state, ) assert not any(case.calls[name] for name in ("models", "capture", "backup", "begin"))