819 lines
34 KiB
Python
819 lines
34 KiB
Python
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()
|