190 lines
8.9 KiB
Python
190 lines
8.9 KiB
Python
from types import SimpleNamespace
|
|
|
|
from backend.app.services.media_codec_dashboard_deployment_reconciliation import (
|
|
MediaCodecDashboardDeploymentReconciliation,
|
|
)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
|
|
class FakePortainer:
|
|
def __init__(self, exit_code=0):
|
|
self.exit_code = exit_code
|
|
self.calls = []
|
|
|
|
def get_container_status(self, name):
|
|
assert name == "gitea-deploy-control"
|
|
return {
|
|
"status": "ok",
|
|
"container": {
|
|
"id": "controller-id",
|
|
"name": "gitea-deploy-control",
|
|
"state": "running",
|
|
},
|
|
}
|
|
|
|
def _config(self):
|
|
return SimpleNamespace(endpoint_id="3")
|
|
|
|
def _request(self, method, path, **kwargs):
|
|
self.calls.append((method, path, kwargs))
|
|
if path.endswith("/containers/controller-id/exec"):
|
|
return FakeResponse({"Id": "exec-1"})
|
|
if path.endswith("/exec/exec-1/start"):
|
|
return FakeResponse({})
|
|
if path.endswith("/exec/exec-1/json"):
|
|
return FakeResponse({"Running": False, "ExitCode": self.exit_code})
|
|
raise AssertionError(path)
|
|
|
|
|
|
def test_embedded_reconciliation_script_is_valid_python():
|
|
fake = FakePortainer()
|
|
MediaCodecDashboardDeploymentReconciliation(fake).run()
|
|
script = fake.calls[0][2]["json"]["Cmd"][2]
|
|
|
|
compile(script, "media-codec-dashboard-reconcile", "exec")
|
|
|
|
|
|
def test_reconciliation_derives_the_verified_live_revision_from_runtime_evidence():
|
|
fake = FakePortainer()
|
|
result = MediaCodecDashboardDeploymentReconciliation(fake).run()
|
|
|
|
assert result == {
|
|
"status": "reconciled",
|
|
"repository": "jens/media-codec-dashboard",
|
|
"container": "Media-Codec-Dashboard",
|
|
}
|
|
|
|
create = fake.calls[0]
|
|
command = create[2]["json"]["Cmd"]
|
|
assert command[:2] == ["python3", "-c"]
|
|
assert command[3:] == [
|
|
"jens/media-codec-dashboard",
|
|
"Media-Codec-Dashboard",
|
|
]
|
|
script = command[2]
|
|
# The mutable registry must identify the protected volume's fixed config.
|
|
assert "/controller-hub-releases/controller-config/projects.json" in script
|
|
assert "/srv/example/appdata/gitea-deploy-runners/controller/projects.json" not in script
|
|
assert "/mnt/cache/appdata/gitea-deploy-runners/state" in script
|
|
assert "/srv/example/appdata/gitea-deploy-runners/deploy.lock" not in script
|
|
assert "/mnt/cache/appdata/gitea-deploy-runners/deploy.lock" in script
|
|
assert "os.path.samefile(registry_path, host_registry_path)" in script
|
|
assert '"/app/data": "/srv/example/appdata/media-codec-dashboard/data"' in script
|
|
assert '"/app/data/quality-runtime": "/srv/example/appdata/tdarr/logs/media-codec-dashboard-quality-audit"' in script
|
|
assert '"/run/secrets": "/srv/example/appdata/media-codec-dashboard/secrets"' in script
|
|
assert '"/tdarr-output": "/srv/example/Media"' in script
|
|
assert '"/tdarr-db"' not in script
|
|
|
|
# Never freeze a one-time recovery helper to a historical deployment SHA.
|
|
assert "971df2cc305c1b447b4feb255d4e9fda871315c3" not in script
|
|
assert "adf53604739d4247193a7b5bfeb8ad8981d9d359" not in script
|
|
assert "live_revision, stale_revision" not in script
|
|
|
|
# The immutable revision must be proven independently by both the running
|
|
# container label and the immutable image label, then resolved in Gitea.
|
|
assert 'container_revision = str(labels.get("org.opencontainers.image.revision") or "")' in script
|
|
assert 'image_revision = str(image_labels.get("org.opencontainers.image.revision") or "")' in script
|
|
assert "container_revision != image_revision" in script
|
|
assert 're.fullmatch(r"[0-9a-f]{40}", live_revision)' in script
|
|
assert 'git_run("rev-parse", f"{live_revision}^{{commit}}")' in script
|
|
assert 'git_run("merge-base", "--is-ancestor", live_revision, main_revision)' in script
|
|
|
|
# The stale controller revision is read from durable state. Recovery is
|
|
# allowed only when that recorded live revision is a real older ancestor of
|
|
# the currently running immutable image, never for arbitrary drift.
|
|
assert 'recorded_live = str(state.get("live_revision") or state.get("revision") or "")' in script
|
|
assert 're.fullmatch(r"[0-9a-f]{40}", recorded_live)' in script
|
|
assert 'git_run("merge-base", "--is-ancestor", recorded_live, live_revision)' in script
|
|
assert 'recorded_live == live_revision' in script
|
|
assert '"live_legacy_provenance": True' in script
|
|
assert create[2]["json"]["AttachStdout"] is False
|
|
assert create[2]["json"]["AttachStderr"] is False
|
|
|
|
|
|
def test_reconciliation_distinguishes_failed_attempt_from_recorded_live_revision():
|
|
fake = FakePortainer()
|
|
MediaCodecDashboardDeploymentReconciliation(fake).run()
|
|
script = fake.calls[0][2]["json"]["Cmd"][2]
|
|
|
|
# Controller schema v3 keeps the real live revision while top-level status
|
|
# describes the most recent deployment attempt. A failed forward attempt is
|
|
# therefore recoverable when the independently verified runtime still equals
|
|
# recorded_live and the failed attempt is a distinct valid descendant.
|
|
assert 'recorded_live_status = str(state.get("live_status") or "")' in script
|
|
assert 'last_attempt = dict(state.get("last_attempt") or {})' in script
|
|
assert 'attempt_revision = str(last_attempt.get("revision") or "")' in script
|
|
assert 'attempt_status = str(last_attempt.get("status") or "")' in script
|
|
assert "current_live_failed_baseline = (" in script
|
|
assert 'recorded_status == "failed"' in script
|
|
assert 'recorded_live_status == "healthy" or current_live_unknown_snapshot_baseline' in script
|
|
assert 'git_run("merge-base", "--is-ancestor", live_revision, attempt_revision)' in script
|
|
assert 'git_run("merge-base", "--is-ancestor", attempt_revision, main_revision)' in script
|
|
|
|
|
|
def test_reconciliation_accepts_exact_current_legacy_snapshot_failure_only_with_matching_artifacts():
|
|
fake = FakePortainer()
|
|
MediaCodecDashboardDeploymentReconciliation(fake).run()
|
|
script = fake.calls[0][2]["json"]["Cmd"][2]
|
|
|
|
assert "current_live_unknown_snapshot_baseline = (" in script
|
|
assert 'recorded_live_status == "unknown"' in script
|
|
assert 'state.get("live_legacy_provenance") is True' in script
|
|
assert 'attempt.get("message") == "Existing production containers require a valid healthy live deployment state"' in script
|
|
assert 'attempt.get("updated_at_epoch") == state.get("updated_at_epoch")' in script
|
|
assert 'current_receipt == {}' in script
|
|
assert 'set(current_artifacts) == {"compose_sha256", "containers"}' in script
|
|
assert 'set(current_artifact_containers) == {expected_container}' in script
|
|
assert 'all(current_stored_artifact.get(field) == container_artifact.get(field) for field in current_immutable_fields)' in script
|
|
assert 'current_live_failed_baseline = (' in script
|
|
assert 'recorded_live_status == "healthy" or current_live_unknown_snapshot_baseline' in script
|
|
|
|
|
|
def test_reconciliation_accepts_only_an_internally_consistent_stale_success_baseline():
|
|
fake = FakePortainer()
|
|
MediaCodecDashboardDeploymentReconciliation(fake).run()
|
|
script = fake.calls[0][2]["json"]["Cmd"][2]
|
|
|
|
# A previously successful controller baseline may legitimately lag behind
|
|
# an independently verified newer immutable runtime. Accept that drift only
|
|
# when the old state is internally self-consistent: healthy live status and
|
|
# a successful last attempt for exactly the recorded old live revision.
|
|
assert "stale_success_baseline = (" in script
|
|
assert 'recorded_status == "success"' in script
|
|
assert 'recorded_live_status == "healthy"' in script
|
|
assert 'attempt_status == "success"' in script
|
|
assert "attempt_revision == recorded_live" in script
|
|
assert 'stale_failed_attempt = recorded_status == "failed"' in script
|
|
assert "if not (stale_success_baseline or stale_failed_attempt):" in script
|
|
assert 'git_run("merge-base", "--is-ancestor", recorded_live, live_revision)' in script
|
|
|
|
|
|
def test_reconciliation_refuses_wrong_live_mount_topology():
|
|
result = MediaCodecDashboardDeploymentReconciliation(FakePortainer(exit_code=48)).run()
|
|
assert result == {
|
|
"status": "validation_failed",
|
|
"reason": "live Media Codec Dashboard mount topology differs from the verified runtime",
|
|
}
|
|
|
|
|
|
def test_reconciliation_refuses_unknown_state_drift():
|
|
result = MediaCodecDashboardDeploymentReconciliation(FakePortainer(exit_code=52)).run()
|
|
assert result == {
|
|
"status": "validation_failed",
|
|
"reason": "controller state drift is not the exact known recoverable state",
|
|
}
|
|
|
|
|
|
def test_reconciliation_refuses_wrong_image_revision_label():
|
|
result = MediaCodecDashboardDeploymentReconciliation(FakePortainer(exit_code=51)).run()
|
|
assert result == {
|
|
"status": "validation_failed",
|
|
"reason": "live Media Codec Dashboard image revision label does not match the verified runtime",
|
|
}
|