from __future__ import annotations import hashlib import importlib.util import json import os import stat import sys import threading from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[2] BROKER_DIR = REPO_ROOT / "ops" / "gitea-deploy-broker" STORE_PATH = BROKER_DIR / "repopulse_adoption_store.py" EXECUTOR_PATH = BROKER_DIR / "repopulse_adoption_executor.py" REPOSITORY = "jens/repopulse" APPLICATION_ID = "repopulse" LEGACY_VOLUME = "repopulse_data" PROTOCOL_VERSION = "itworx.repopulse-legacy-adoption/v1" def _load_module(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module def _sha(value: str) -> str: return hashlib.sha256(value.encode("ascii")).hexdigest() def _canonical_sha(value: object) -> str: payload = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ).encode("ascii") return hashlib.sha256(payload).hexdigest() def _preflight_projection() -> dict: return { "repository": REPOSITORY, "main_revision": "a" * 40, "live_revision": "b" * 40, "container_id_sha256": _sha("legacy-container"), "container_image_sha256": _sha("legacy-image"), "queue": {"queued": 0, "active": 0, "completed": 572, "failed": 0}, "database": { "database_bytes": 44_576_768, "journal_bytes": 0, "scan_count": 572, "compacted_scan_count": 96, }, "container_state": "running", "container_health": "healthy", } def _command(executor): idempotency_sha256 = _sha(f"{PROTOCOL_VERSION}:{REPOSITORY}:repopulse-adoption-2026") projection = _preflight_projection() return executor.AdoptionCommand( intent_id=f"rpa-{idempotency_sha256[:20]}", idempotency_sha256=idempotency_sha256, preflight_sha256=_canonical_sha(projection), preflight_projection=projection, ) def _target_gates() -> dict: return { "profileId": "repopulse-authenticated", "profileVersion": "enrollment-profile/v1", "planSha256": _sha("reviewed fixed profile plan"), "profileReady": True, "controllerTrustReady": True, "controllerBootstrapSecretImported": True, "typedEnrollmentTrustReady": True, "targetSecretReady": True, "exactPolicyReady": True, "authenticatedRouteReady": True, "accessListReady": True, "mutationBoundaryReady": True, "managedRuntimeReady": True, "scanSourceMatches": True, "scanMountReadOnly": True, "scanPermissionPolicyPersistent": True, "scanAccessForRuntimeIdentityReady": True, "ownedDataTargetReady": True, "managedEnvironmentReady": True, "legacyEnvironmentRetained": True, "maintenanceAuthReady": True, "controllerAdoptionCapabilityReady": True, "controllerRegistryWriterReady": True, "controllerSecretWriterReady": True, } def _quiescence(command, binding: str) -> dict: projection = command.preflight_projection return { "intentId": command.intent_id, "globalDeployLockHeld": True, "projectLockHeld": True, "maintenanceModeActive": True, "queueFrozen": True, "queuedJobs": 0, "activeJobs": 0, "maintenanceModeAuthenticated": True, "containerName": "repopulse", "containerIdSha256": projection["container_id_sha256"], "containerImageSha256": projection["container_image_sha256"], "containerState": "exited", "legacyVolume": LEGACY_VOLUME, "controllerIntentDurable": True, "controllerIntentBindingSha256": binding, } def _snapshot(command, binding: str) -> dict: return { "intentId": command.intent_id, "globalDeployLockHeld": True, "projectLockHeld": True, "legacyContainerStopped": True, "legacyVolume": LEGACY_VOLUME, "legacyVolumeRetained": True, "sqliteBackupMethod": "sqlite_online_backup", "quickCheck": "ok", "integrityCheck": "ok", "sourceLogicalSha256": _sha("logical-sqlite-data"), "snapshotLogicalSha256": _sha("logical-sqlite-data"), "snapshotFileSha256": _sha("fresh-sqlite-backup"), "sourceScanCount": 572, "snapshotScanCount": 572, "sourceCompactedScanCount": 96, "snapshotCompactedScanCount": 96, "sourceQueueSha256": _sha("queue-state"), "snapshotQueueSha256": _sha("queue-state"), "sourceAuxiliaryTreeSha256": _sha("auxiliary-tree"), "sourceAuxiliaryFileCount": 42, "targetDataPathWasAbsent": True, "snapshotReceiptDurable": True, "legacyEnvironmentRetained": True, "managedEnvironmentFilePrivate": True, "controllerIntentBindingSha256": binding, } def _copy_evidence(command, snapshot: dict) -> dict: return { "intentId": command.intent_id, "snapshotFileSha256": snapshot["snapshotFileSha256"], "targetLogicalSha256": snapshot["snapshotLogicalSha256"], "targetScanCount": snapshot["snapshotScanCount"], "targetCompactedScanCount": snapshot["snapshotCompactedScanCount"], "targetQueueSha256": snapshot["snapshotQueueSha256"], "sourceAuxiliaryTreeSha256": snapshot["sourceAuxiliaryTreeSha256"], "targetAuxiliaryTreeSha256": snapshot["sourceAuxiliaryTreeSha256"], "sourceAuxiliaryFileCount": snapshot["sourceAuxiliaryFileCount"], "targetAuxiliaryFileCount": snapshot["sourceAuxiliaryFileCount"], "targetQuickCheck": "ok", "targetIntegrityCheck": "ok", "targetDataUid": 10001, "targetDataGid": 10001, "targetRecoveryStatus": "healthy", "managedDataMountMatches": True, "copySourceIsVerifiedSnapshot": True, "legacyVolumeRetained": True, "legacyEnvironmentRetained": True, "managedEnvironmentFilePrivate": True, } def _legacy_rollback_evidence(command, legacy_projects_sha256: str) -> dict: return { "intentId": command.intent_id, "repository": REPOSITORY, "legacyContainerName": "repopulse", "legacyVolume": LEGACY_VOLUME, "legacyProjectsSha256": legacy_projects_sha256, "rollbackSnapshotSha256": _sha("durable old compose rollback snapshot"), "rollbackSnapshotDurable": True, "legacyVolumeRetained": True, "legacyEnvironmentRetained": True, } def _deployment_receipt(command, snapshot: dict, copy: dict) -> dict: revision = command.preflight_projection["main_revision"] return { "repository": REPOSITORY, "revision": revision, "containerRevision": revision, "containerIdSha256": _sha("managed-container"), "deploymentReceiptSchema": "gitea-deploy-receipt.v1", "deploymentReceiptSha256": _sha("verified-deploy-receipt"), "deploymentReceiptVerified": True, "runtimeHealthy": True, "publishedHostPorts": [], "managedDataMountMatches": copy["managedDataMountMatches"], "targetDataUid": copy["targetDataUid"], "targetDataGid": copy["targetDataGid"], "targetLogicalSha256": copy["targetLogicalSha256"], "targetScanCount": copy["targetScanCount"], "targetCompactedScanCount": copy["targetCompactedScanCount"], "targetQueueSha256": copy["targetQueueSha256"], "targetAuxiliaryTreeSha256": copy["targetAuxiliaryTreeSha256"], "targetAuxiliaryFileCount": copy["targetAuxiliaryFileCount"], "targetQuickCheck": copy["targetQuickCheck"], "targetIntegrityCheck": copy["targetIntegrityCheck"], "targetRecoveryStatus": copy["targetRecoveryStatus"], "targetQueuedJobs": 0, "targetActiveJobs": 0, "legacyVolume": LEGACY_VOLUME, "legacyVolumeRetained": True, "legacyEnvironmentRetained": True, "managedEnvironmentFilePrivate": True, "scanSourceMatches": True, "scanMountReadOnly": True, "scanPermissionPolicyPersistent": True, "scanAccessForRuntimeIdentityReady": True, "containerName": "repopulse", "containerImageSha256": _sha("managed-image"), "snapshotFileSha256": snapshot["snapshotFileSha256"], } @pytest.fixture def executor_env(tmp_path, monkeypatch): module = _load_module("repopulse_adoption_store", STORE_PATH) controller = tmp_path / "controller" controller.mkdir() controller.chmod(0o755) (controller / "projects.json").write_text( json.dumps( { REPOSITORY: { "branch": "main", "compose_files": ["docker-compose.yml"], "compose_project": "repopulse", "containers": ["repopulse"], "env_source": "/srv/example/appdata/repopulse/deploy.env", "exact_source": True, "gitea_repository_id": 64, "gitea_webhook_secret_file": "jens__repopulse", "health_timeout": 180, "slug": "repopulse", "source_dir": "/mnt/cache/appdata/gitea-deploy-runners/sources/repopulse", "validation": { "contexts": ["Managed validation / full (push)"], "max_poll_interval_seconds": 30, "poll_interval_seconds": 5, "required": True, "timeout_seconds": 1800, "validation_gate_version": 1, }, "verify_oci_revision": False, }, "jens/other": {"slug": "other"}, }, sort_keys=True, ) + "\n", encoding="utf-8", ) (controller / "projects.json").chmod(0o600) secrets = controller / "appops-deployment-secrets" secrets.mkdir() secrets.chmod(0o700) state = tmp_path / "state" state.mkdir() state.chmod(0o755) locks = state / "locks" locks.mkdir() locks.chmod(0o755) global_lock = tmp_path / "deploy.lock" state_store = state / "repopulse-adoption" monkeypatch.setattr(module, "CONTROLLER_READONLY_ROOT", controller) monkeypatch.setattr(module, "CONTROLLER_RW_ROOT", controller) monkeypatch.setattr(module, "STATE_ROOT", state) monkeypatch.setattr(module, "GLOBAL_DEPLOY_LOCK", global_lock) monkeypatch.setattr(module, "PROJECT_LOCK", locks / "repopulse.lock") monkeypatch.setattr(module, "ADOPTION_STATE_ROOT", state_store) monkeypatch.setattr(module, "EXECUTION_ENABLED", True) monkeypatch.setattr(module, "_IS_POSIX_CONTROLLER", True) test_uid = controller.stat().st_uid monkeypatch.setattr(module, "_ROOT_UID", test_uid) monkeypatch.setattr(module, "_effective_uid", lambda: test_uid) if os.name != "posix": monkeypatch.setattr( module, "_mode_bits", lambda path, info: ( 0o700 if path.name in {"appops-deployment-secrets", "repopulse-adoption"} else 0o755 if stat.S_ISDIR(info.st_mode) else 0o600 ), ) monkeypatch.setattr(module, "_directory_fsync", lambda _path: None) executor = _load_module("repopulse_adoption_executor_test", EXECUTOR_PATH) return module, module.RepoPulseAdoptionStore(), executor, controller, state_store class Effects: def __init__( self, store_module, command, *, bad_receipt_field: str | None = None, bad_copy=False, bad_validation=False, fail_rollback_stop=False, bad_live_completion=False, ): self.store_module = store_module self.command = command self.bad_receipt_field = bad_receipt_field self.bad_copy = bad_copy self.bad_validation = bad_validation self.fail_rollback_stop = fail_rollback_stop self.bad_live_completion = bad_live_completion self.events: list[str] = [] self.credential = b"target-secret-never-log-" + b"x" * 40 def _event(self, name: str) -> None: assert self.store_module._transaction_owner(self.store_module._lock_identity()) == threading.get_ident() self.events.append(name) def target_gates(self): self._event("gates") return _target_gates() def capture_legacy_rollback(self, *, intent_id: str, expected_legacy_projects_sha256: str): self._event("capture_legacy_rollback") assert intent_id == self.command.intent_id return _legacy_rollback_evidence(self.command, expected_legacy_projects_sha256) def ensure_quiesced( self, *, intent_id: str, intent_binding_sha256: str, expected_main_revision: str, expected_container_id_sha256: str, expected_container_image_sha256: str, ): self._event("quiesce") assert intent_id == self.command.intent_id assert expected_main_revision == self.command.preflight_projection["main_revision"] assert expected_container_id_sha256 == self.command.preflight_projection["container_id_sha256"] assert expected_container_image_sha256 == self.command.preflight_projection["container_image_sha256"] return _quiescence(self.command, intent_binding_sha256) def create_or_verify_fresh_snapshot( self, *, intent_id: str, intent_binding_sha256: str, expected_main_revision: str, expected_snapshot, ): self._event("snapshot") assert intent_id == self.command.intent_id assert expected_main_revision == self.command.preflight_projection["main_revision"] proof = _snapshot(self.command, intent_binding_sha256) if expected_snapshot is not None: assert expected_snapshot["fileSha256"] == proof["snapshotFileSha256"] return proof def copy_and_verify_managed_data( self, *, intent_id: str, expected_main_revision: str, snapshot_evidence: dict ): self._event("copy") assert intent_id == self.command.intent_id assert expected_main_revision == self.command.preflight_projection["main_revision"] proof = _copy_evidence(self.command, snapshot_evidence) if self.bad_copy: proof["copySourceIsVerifiedSnapshot"] = False return proof def registration_material(self, *, intent_id: str): self._event("registration_material") assert intent_id == self.command.intent_id return "enr-0123456789abcdef0123", self.credential def validate_exact_sha_ci_locked(self, *, repository: str, revision: str): self._event("validation_gate") assert repository == REPOSITORY assert revision == self.command.preflight_projection["main_revision"] _raw, projects, credential = self.store_module._check_controller_aliases() assert projects[REPOSITORY]["managed_profile"] == "repopulse-authenticated" assert credential == self.credential return { "repository": REPOSITORY, "revision": revision, "contexts": {"Managed validation / full (push)": "failure" if self.bad_validation else "success"}, } def deploy_exact_sha_locked( self, *, intent_id: str, revision: str, controller_registry_receipt: dict, snapshot_evidence: dict, copied_data_evidence: dict, ): self._event("deploy") assert intent_id == self.command.intent_id assert revision == self.command.preflight_projection["main_revision"] assert controller_registry_receipt["phase"] == "registry_committed" proof = _deployment_receipt(self.command, snapshot_evidence, copied_data_evidence) if self.bad_receipt_field: proof[self.bad_receipt_field] = "wrong" if self.bad_receipt_field in {"repository", "revision", "containerRevision"} else False return proof def stop_managed_writer(self, *, intent_id: str, revision: str): self._event("stop_managed_writer") assert intent_id == self.command.intent_id assert revision == self.command.preflight_projection["main_revision"] return { "intentId": intent_id, "repository": REPOSITORY, "containerName": "repopulse", "managedWriterStopped": not self.fail_rollback_stop, "activeWriterCount": 0 if not self.fail_rollback_stop else 1, "legacyVolumeRetained": True, } def restore_legacy_registry( self, *, intent_id: str, expected_legacy_projects_sha256: str, controller_registry_receipt: dict | None, ): self._event("restore_legacy_registry") assert intent_id == self.command.intent_id assert expected_legacy_projects_sha256 == _legacy_rollback_evidence(self.command, expected_legacy_projects_sha256)["legacyProjectsSha256"] projects_path = self.store_module.CONTROLLER_RW_ROOT / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects[REPOSITORY] = self.store_module._legacy_registry_record() projects_path.write_text(json.dumps(projects, sort_keys=True) + "\n", encoding="utf-8") projects_path.chmod(0o600) return { "intentId": intent_id, "repository": REPOSITORY, "projectsSha256": expected_legacy_projects_sha256, "legacyRegistryRestored": True, "legacyVolumeRetained": True, } def restore_legacy_runtime( self, *, legacy_rollback_evidence: dict, legacy_registry_evidence: dict, failed_revision: str, ): self._event("restore_legacy_runtime") assert failed_revision == self.command.preflight_projection["main_revision"] assert legacy_registry_evidence["legacyRegistryRestored"] is True assert legacy_rollback_evidence["legacyVolume"] == LEGACY_VOLUME return { "intentId": self.command.intent_id, "repository": REPOSITORY, "containerName": "repopulse", "projectsSha256": legacy_registry_evidence["projectsSha256"], "legacyRuntimeHealthy": True, "activeWriterCount": 1, "legacyVolumeRetained": True, "legacyEnvironmentRetained": True, } def verify_live_completion_locked(self, *, completion: dict): self._event("verify_completion") proof = { "repository": REPOSITORY, "revision": completion["mainRevision"], "containerRevision": completion["mainRevision"], "deploymentReceiptSha256": completion["deploymentReceiptSha256"], "containerIdSha256": completion["containerIdSha256"], "containerImageSha256": completion["containerImageSha256"], "runtimeHealthy": True, "legacyVolume": LEGACY_VOLUME, "legacyVolumeRetained": True, } if self.bad_live_completion: proof["containerRevision"] = "f" * 40 return proof def test_executor_stays_hard_disabled_without_touching_store_or_effects(executor_env, monkeypatch): store_module, store, executor_module, _controller, state_store = executor_env monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", False) command = _command(executor_module) effects = object() executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) assert executor_module.EXECUTION_ENABLED is False assert executor_module.execution_status()["hasEffectfulRoute"] is False with pytest.raises(executor_module.AdoptionExecutorError) as error: executor.execute(command) assert error.value.code == "controller_adoption_execution_disabled" assert not state_store.exists() assert store_module._transaction_owner(store_module._lock_identity()) is None def test_executor_holds_project_then_global_transaction_through_deploy_receipt(executor_env, monkeypatch): store_module, store, executor_module, controller, state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) result = executor.execute(command) assert effects.events == ["gates", "capture_legacy_rollback", "quiesce", "snapshot", "copy", "registration_material", "validation_gate", "deploy"] assert result["repository"] == REPOSITORY assert result["applicationId"] == APPLICATION_ID assert result["deploymentReceipt"]["revision"] == command.preflight_projection["main_revision"] assert result["deploymentReceipt"]["containerRevision"] == command.preflight_projection["main_revision"] assert result["legacyVolumeRetained"] is True intent = store.load_intent() assert intent is not None assert intent["phase"] == "snapshot_verified" assert intent["status"] == "in_progress" assert intent["registration"]["status"] == "committed" assert store.load_receipt()["phase"] == "registry_committed" persisted = (state_store / "intent.json").read_text(encoding="utf-8") persisted += (state_store / "receipt.json").read_text(encoding="utf-8") assert effects.credential.decode("ascii") not in persisted assert effects.credential.decode("ascii") not in json.dumps(result) assert result["completion"]["legacyVolumeRetained"] is True assert result["completion"]["mainRevision"] == command.preflight_projection["main_revision"] completion_path = state_store / "completion.json" assert completion_path.is_file() assert executor_module._store._mode_bits(completion_path, completion_path.stat()) == 0o600 assert json.loads((controller / "projects.json").read_text(encoding="utf-8"))[REPOSITORY]["managed_profile"] == "repopulse-authenticated" def test_completion_replay_reverifies_registry_and_live_receipt_without_redeploy(executor_env, monkeypatch): store_module, store, executor_module, _controller, _state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) executor.execute(command) effects.events.clear() replay = executor.execute(command) assert effects.events == ["verify_completion"] assert replay["phase"] == "completion_verified" assert replay["completion"]["intentBindingSha256"] == store.load_intent()["intentBindingSha256"] def test_completion_replay_mismatch_fails_closed_without_starting_or_deploying(executor_env, monkeypatch): store_module, store, executor_module, _controller, _state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) executor.execute(command) effects.events.clear() effects.bad_live_completion = True with pytest.raises(executor_module.AdoptionExecutorError) as error: executor.execute(command) assert error.value.code == "controller_completion_live_receipt_invalid" assert effects.events == ["verify_completion"] assert store.load_intent()["status"] == "recovery_required" def test_registry_cutover_without_completion_requires_reconciliation(executor_env, monkeypatch): store_module, store, executor_module, _controller, state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) executor.execute(command) (state_store / "completion.json").unlink() effects.events.clear() with pytest.raises(executor_module.AdoptionExecutorError) as error: executor.execute(command) assert error.value.code == "controller_completion_reconciliation_required" assert effects.events == [] assert store.load_intent()["status"] == "recovery_required" def test_private_legacy_snapshot_is_durable_and_restores_only_fixed_compose_runtime(executor_env, tmp_path, monkeypatch): store_module, store, executor_module, controller, state_store = executor_env legacy_env = tmp_path / "legacy-deploy.env" legacy_env.write_text("PRIVATE=not-read\n", encoding="utf-8") legacy_env.chmod(0o600) fixed_legacy = store_module._legacy_registry_record() fixed_legacy["env_source"] = str(legacy_env) monkeypatch.setattr(store_module, "_legacy_registry_record", lambda: dict(fixed_legacy)) projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects[REPOSITORY] = fixed_legacy projects_path.write_text(json.dumps(projects, sort_keys=True) + "\n", encoding="utf-8") projects_path.chmod(0o600) monkeypatch.setattr(executor_module, "LEGACY_ENVIRONMENT_FILE", legacy_env) class FakeDeployCore: def __init__(self): self.restored = [] def repository_path(self, repository, project): assert repository == REPOSITORY assert project["compose_project"] == "repopulse" return tmp_path / "repopulse.git" def inspect_container(self, name): assert name == "repopulse" return { "Id": "c" * 64, "Image": "sha256:" + "d" * 64, "State": {"Running": True}, "Mounts": [{"Type": "volume", "Name": LEGACY_VOLUME, "RW": True}], } def output(self, command): assert command == [ "docker", "ps", "--all", "--filter", f"volume={LEGACY_VOLUME}", "--format", "{{.Names}}", ] return "repopulse\n" def capture_compose_rollback(self, project): assert project["_repository_name"] == REPOSITORY assert project["compose_files"] == ["docker-compose.yml"] return { "revision": "b" * 40, "legacy_provenance": True, "receipt": {}, "compose_containers": ["repopulse"], "containers": [{ "name": "repopulse", "container_id": "c" * 64, "image_id": "sha256:" + "d" * 64, "image_ref": "repopulse:legacy", "rollback_ref": "gitea-deploy-rollback/repopulse-bbbbbbbbbbbb", "was_running": True, }], "captured_at_epoch": 1, } def restore_compose_snapshot(self, project, snapshot, repository, failed_revision): assert project["compose_files"] == ["docker-compose.yml"] assert project["env_source"] == str(legacy_env) assert snapshot["legacy_provenance"] is True assert repository == REPOSITORY assert failed_revision == "a" * 40 self.restored.append("legacy-compose") def verify_containers(self, project, *, expected_revision, expected_image_ids): assert project["compose_project"] == "repopulse" assert expected_revision is None assert expected_image_ids == {"repopulse": "sha256:" + "d" * 64} return {"repopulse": {"image_id": "sha256:" + "d" * 64}} def container_state(self, name): assert name == "repopulse" return True, "healthy", "repopulse" core = FakeDeployCore() adapter = executor_module.FixedLegacyRollbackAdapter(store=store, deployment_core=core) intent_id = "rpa-0123456789abcdef0123" projects_sha = hashlib.sha256(projects_path.read_bytes()).hexdigest() with store.transaction(): evidence = adapter.capture_legacy_rollback( intent_id=intent_id, expected_legacy_projects_sha256=projects_sha, ) snapshot_file = state_store / f"legacy-runtime-{intent_id}.json" assert snapshot_file.is_file() assert executor_module._store._mode_bits(snapshot_file, snapshot_file.stat()) == 0o600 raw_record = json.loads(snapshot_file.read_text(encoding="utf-8")) assert raw_record["snapshot"]["containers"][0]["name"] == "repopulse" proof = adapter.restore_legacy_runtime( legacy_rollback_evidence=evidence, legacy_registry_evidence={ "intentId": intent_id, "repository": REPOSITORY, "projectsSha256": projects_sha, "legacyRegistryRestored": True, "legacyVolumeRetained": True, }, failed_revision="a" * 40, ) assert core.restored == ["legacy-compose"] assert proof["legacyRuntimeHealthy"] is True assert proof["activeWriterCount"] == 1 assert "legacy-deploy.env" not in json.dumps(proof) @pytest.mark.parametrize("field", ["repository", "revision", "containerRevision"]) def test_unverified_exact_sha_controller_receipt_enters_durable_recovery(executor_env, monkeypatch, field): store_module, store, executor_module, _controller, state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command, bad_receipt_field=field) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) with pytest.raises(executor_module.AdoptionExecutorError) as error: executor.execute(command) assert error.value.code == "controller_deployment_receipt_mismatch" intent = store.load_intent() assert intent is not None assert intent["phase"] == "snapshot_verified" assert intent["status"] == "recovery_required" assert intent["recovery"]["reason"] == "target_health_failed" assert LEGACY_VOLUME in str(intent) assert "target-secret-never-log" not in str(error.value) assert effects.events[-4:] == ["deploy", "stop_managed_writer", "restore_legacy_registry", "restore_legacy_runtime"] assert effects.events.index("stop_managed_writer") < effects.events.index("restore_legacy_registry") < effects.events.index("restore_legacy_runtime") assert (state_store / "receipt.json").is_file() def test_unverified_managed_copy_blocks_registry_cas_and_records_recovery(executor_env, monkeypatch): store_module, store, executor_module, controller, _state_store = executor_env command = _command(executor_module) effects = Effects(store_module, command, bad_copy=True) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) executor = executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects) with pytest.raises(executor_module.AdoptionExecutorError) as error: executor.execute(command) assert error.value.code == "managed_data_copy_not_verified" assert effects.events == ["gates", "capture_legacy_rollback", "quiesce", "snapshot", "copy", "stop_managed_writer", "restore_legacy_registry", "restore_legacy_runtime"] assert json.loads((controller / "projects.json").read_text(encoding="utf-8"))[REPOSITORY]["env_source"] == "/srv/example/appdata/repopulse/deploy.env" assert not (controller / "appops-deployment-secrets" / "jens__repopulse").exists() intent = store.load_intent() assert intent is not None assert intent["phase"] == "quiesced" assert intent["status"] == "recovery_required" assert intent["recovery"]["reason"] == "database_backup_failed" def test_exact_sha_ci_gate_is_required_before_deploy_and_failed_gate_restores_legacy(executor_env, monkeypatch): _store_module, store, executor_module, controller, _state_store = executor_env command = _command(executor_module) effects = Effects(_store_module, command, bad_validation=True) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) with pytest.raises(executor_module.AdoptionExecutorError) as error: executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects).execute(command) assert error.value.code == "exact_sha_validation_not_passed" assert effects.events[-3:] == ["stop_managed_writer", "restore_legacy_registry", "restore_legacy_runtime"] assert "deploy" not in effects.events assert json.loads((controller / "projects.json").read_text(encoding="utf-8"))[REPOSITORY]["env_source"] == "/srv/example/appdata/repopulse/deploy.env" assert store.load_intent()["status"] == "recovery_required" def test_legacy_runtime_is_never_restarted_when_managed_writer_stop_is_unproven(executor_env, monkeypatch): _store_module, store, executor_module, _controller, _state_store = executor_env command = _command(executor_module) effects = Effects(_store_module, command, bad_receipt_field="revision", fail_rollback_stop=True) monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) with pytest.raises(executor_module.AdoptionExecutorError) as error: executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects).execute(command) assert error.value.code == "legacy_rollback_incomplete" assert effects.events[-1] == "stop_managed_writer" assert "restore_legacy_registry" not in effects.events assert "restore_legacy_runtime" not in effects.events assert store.load_intent()["status"] == "recovery_required" def test_invalid_or_changed_fixed_preflight_is_rejected_before_external_effects(executor_env, monkeypatch): _store_module, store, executor_module, _controller, state_store = executor_env command = _command(executor_module) changed_projection = dict(command.preflight_projection) changed_projection["repository"] = "jens/other" changed = executor_module.AdoptionCommand( intent_id=command.intent_id, idempotency_sha256=command.idempotency_sha256, preflight_sha256=command.preflight_sha256, preflight_projection=changed_projection, ) effects = object() monkeypatch.setattr(executor_module, "EXECUTION_ENABLED", True) with pytest.raises(executor_module.AdoptionExecutorError) as error: executor_module.RepoPulseAdoptionExecutor(store=store, effects=effects).execute(changed) assert error.value.code == "preflight_projection_invalid" assert not state_store.exists()