199 lines
11 KiB
Python
199 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from test_deploy_broker import (
|
|
_appops_known_legacy_state,
|
|
_appops_rollback_intent_fixture,
|
|
deploy_module as deploy_module,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def bridge(deploy_module, monkeypatch, tmp_path):
|
|
m = deploy_module
|
|
revision = "c" * 40
|
|
state_root = tmp_path / "state"
|
|
state_root.mkdir(mode=0o700)
|
|
monkeypatch.setattr(m, "STATE_ROOT", state_root)
|
|
monkeypatch.setattr(m, "APPOPS_LEGACY_INTENT_ROOT", state_root / "appops-legacy")
|
|
monkeypatch.setattr(m, "APPOPS_LEGACY_INTENT_PATH", state_root / "appops-legacy" / "unraid-appops-gateway.json")
|
|
prior = _appops_known_legacy_state()
|
|
monkeypatch.setattr(m, "APPOPS_LEGACY_ATTEMPT_METADATA_SHA256", m._canonical_json_sha256(prior["last_attempt"]["metadata"]))
|
|
monkeypatch.setattr(m, "APPOPS_LEGACY_PROJECTION_SHA256", m._canonical_json_sha256(m._appops_legacy_state_projection(prior)))
|
|
repository, metadata, source, green, backup, sample = _appops_rollback_intent_fixture(m, revision, "e" * 64)
|
|
project = {"slug": "unraid-appops-gateway", "compose_project": "unraid-appops-gateway",
|
|
"_repository_name": repository, "branch": "main", "gitea_repository_id": 9,
|
|
"containers": ["Unraid-AppOps-Gateway"], "compose_files": ["docker-compose.yml"],
|
|
"verify_oci_revision": True, "environment": {"APPOPS_DATA_PATH": str(m.APPOPS_DATA_ROOT)}}
|
|
runtime = {"containerId": m.APPOPS_LEGACY_OLD_CONTAINER_ID, "imageId": m.APPOPS_LEGACY_OLD_IMAGE_ID,
|
|
"imageRef": m.APPOPS_LEGACY_OLD_IMAGE_REF, "sourceRepository": repository}
|
|
artifacts = m._appops_legacy_state_override(prior, runtime)["live_artifacts"]
|
|
artifacts["compose_sha256"] = ""
|
|
monkeypatch.setattr(m, "require_host_root", lambda: None)
|
|
monkeypatch.setattr(m, "_appops_recovery_config_unchanged", lambda _p: None)
|
|
monkeypatch.setattr(m, "validate_commit", lambda _r, _b, rev: rev)
|
|
monkeypatch.setattr(m, "validate_recorded_receipt", lambda *_a: None)
|
|
monkeypatch.setattr(m, "_appops_recovery_identity", lambda *_a: runtime.copy())
|
|
monkeypatch.setattr(m, "deployment_container_artifacts", lambda _p: copy.deepcopy(artifacts["containers"]))
|
|
monkeypatch.setattr(m, "deployment_artifacts", lambda _p: copy.deepcopy(artifacts))
|
|
monkeypatch.setattr(m, "verify_appops_green_delivery", lambda rev: green if rev == revision else pytest.fail("wrong candidate"))
|
|
monkeypatch.setattr(m, "verify_appops_legacy_source", lambda *_a: source)
|
|
monkeypatch.setattr(m, "revalidate_deployment_candidate", lambda *_a: None)
|
|
proof = sample["migrationProof"]["runtime"]
|
|
monkeypatch.setattr(m, "appops_legacy_compose_model", lambda *_a: ({"name": project["compose_project"]}, proof["resolvedComposeSha256"], proof["resolvedEnvironmentSha256"]))
|
|
monkeypatch.setattr(m, "_appops_legacy_inspect_runtime", lambda *_a, **_k: ({}, {}, runtime, None))
|
|
monkeypatch.setattr(m, "appops_legacy_runtime_hashes", lambda *_a: proof)
|
|
events = []
|
|
monkeypatch.setattr(m, "capture_appops_data_backup", lambda *_a: events.append("backup") or copy.deepcopy(backup))
|
|
monkeypatch.setattr(m, "verify_appops_data_backup", lambda *_a: None)
|
|
monkeypatch.setattr(m, "capture_compose_rollback", lambda *_a, **_k: events.append("snapshot") or copy.deepcopy(sample["rollbackSnapshot"]))
|
|
monkeypatch.setattr(m, "_appops_legacy_tag_snapshot", lambda *_a: events.append("tag"))
|
|
path = state_root / "unraid-appops-gateway.json"
|
|
path.write_text(json.dumps(prior), encoding="utf-8")
|
|
path.chmod(0o644)
|
|
return m, project, tmp_path, revision, metadata, path, events
|
|
|
|
|
|
def prepare_bridge(bridge):
|
|
m, project, repo, revision, metadata, _path, _events = bridge
|
|
state, raw, digest = m.read_appops_state_preimage(project)
|
|
return m.prepare_appops_legacy_migration(project, project["_repository_name"], repo, revision, metadata, state, raw, digest)
|
|
|
|
|
|
def test_reattested_state_prepares_durable_migration_and_completed_rollback_retry(bridge):
|
|
m, project, repo, revision, metadata, path, events = bridge
|
|
m.reattest_appops_live_state(project, repo, m.APPOPS_LEGACY_OLD_REVISION)
|
|
state, raw, digest = m.read_appops_state_preimage(project)
|
|
migration = prepare_bridge(bridge)
|
|
assert events == ["backup", "snapshot", "tag"]
|
|
intent, intent_hash = m.read_appops_legacy_intent()
|
|
assert intent["stateSha256"] == digest
|
|
assert intent["migrationProof"]["recovery"]["auditSha256"] == m._canonical_json_sha256(state["recovery"])
|
|
assert m._appops_legacy_pending_state_is_bound(state, raw, digest, intent, revision, project, repo)
|
|
wrong_intent = copy.deepcopy(intent)
|
|
wrong_intent["migrationProof"]["recovery"]["auditSha256"] = "f" * 64
|
|
assert not m._appops_legacy_pending_state_is_bound(state, raw, digest, wrong_intent, revision, project, repo)
|
|
|
|
project["_compose_rollback"] = migration["snapshot"]
|
|
project["_rollback_restored_revision"] = m.APPOPS_LEGACY_OLD_REVISION
|
|
m.write_state(project["_repository_name"], project, m.APPOPS_LEGACY_OLD_REVISION, "success",
|
|
m.APPOPS_LEGACY_ROLLBACK_STATE_MESSAGE, live_legacy_provenance=True,
|
|
live_receipt={}, rollback_snapshot={}, attempt_revision=revision,
|
|
attempt_metadata={"operation": "rollback", "from_revision": revision}, expected_state_sha256=digest)
|
|
restored, restored_raw, restored_hash = m.read_appops_state_preimage(project)
|
|
assert m._appops_legacy_rollback_state_matches(restored, project, revision, intent)
|
|
tampered = copy.deepcopy(restored)
|
|
tampered["recovery"]["at_epoch"] += 1
|
|
assert not m._appops_legacy_rollback_state_matches(tampered, project, revision, intent)
|
|
completed = m._appops_legacy_rebind_observed_state(intent, restored_hash)
|
|
completed["status"] = "rolled_back"
|
|
m.write_appops_legacy_intent(completed, expected_sha256=intent_hash)
|
|
assert m.appops_legacy_rollback_state_anchor(restored, restored_raw, completed, revision)
|
|
events.clear()
|
|
retried = prepare_bridge(bridge)
|
|
assert events == ["tag"]
|
|
assert retried["intent"]["migrationProof"] == intent["migrationProof"]
|
|
assert retried["intent"]["status"] == "deploying"
|
|
|
|
|
|
def test_candidate_crash_reconciles_the_reattested_preimage_to_a_real_receipt(bridge, monkeypatch):
|
|
m, project, repo, revision, metadata, path, _events = bridge
|
|
m.reattest_appops_live_state(project, repo, m.APPOPS_LEGACY_OLD_REVISION)
|
|
original, raw, digest = m.read_appops_state_preimage(project)
|
|
prepare_bridge(bridge)
|
|
receipt = {"repository": project["_repository_name"], "revision": revision,
|
|
"tree": metadata["tree"], "build": metadata}
|
|
monkeypatch.setattr(m, "verify_deployment", lambda *_a: receipt)
|
|
result = m.reconcile_appops_legacy_candidate(
|
|
project, project["_repository_name"], repo, revision, metadata, original, raw, digest,
|
|
)
|
|
assert result == receipt
|
|
state = json.loads(path.read_bytes())
|
|
assert state["live_revision"] == revision
|
|
assert state["live_receipt"] == receipt
|
|
assert state["live_legacy_provenance"] is False
|
|
assert state["recovery"] == original["recovery"]
|
|
assert m.read_appops_legacy_intent()[0]["status"] == "deployed"
|
|
|
|
|
|
@pytest.mark.parametrize("damage", ["backup-bytes", "backup-mode", "backup-link", "audit", "state"])
|
|
def test_reattest_bridge_rejects_tampering_before_backup_or_docker(bridge, damage):
|
|
m, project, repo, _revision, _metadata, path, events = bridge
|
|
result = m.reattest_appops_live_state(project, repo, m.APPOPS_LEGACY_OLD_REVISION)
|
|
backup = path.parent / result["backupFile"]
|
|
if damage == "backup-bytes":
|
|
backup.write_bytes(b"{}")
|
|
elif damage == "backup-mode":
|
|
backup.chmod(0o644)
|
|
elif damage == "backup-link":
|
|
target = backup.with_suffix(".saved")
|
|
backup.rename(target)
|
|
backup.symlink_to(target)
|
|
else:
|
|
state = json.loads(path.read_bytes())
|
|
if damage == "audit":
|
|
state["recovery"]["observed_image_id"] = "sha256:" + "f" * 64
|
|
else:
|
|
state["live_artifacts"]["containers"]["Unraid-AppOps-Gateway"]["image_id"] = "sha256:" + "f" * 64
|
|
path.write_text(json.dumps(state))
|
|
with pytest.raises(m.AppOpsLegacyNoMutation):
|
|
prepare_bridge(bridge)
|
|
assert events == []
|
|
assert not m.APPOPS_LEGACY_INTENT_PATH.exists()
|
|
|
|
|
|
def prior_record(m, revision):
|
|
semantic = m._canonical_json_sha256({"gitea_repository_id": 9, "ref": "refs/heads/main", "revision": revision})
|
|
record = {"schema": "gitea-webhook-inbox.v2", "semantic_sha256": semantic,
|
|
"body_sha256": "a" * 64, "delivery": "12345678-1234-4234-8234-123456789012",
|
|
"gitea_repository_id": 9, "ref": "refs/heads/main", "repository": "jens/unraid-appops-gateway",
|
|
"revision": revision, "source": "gitea_webhook", "status": "failed", "reason": "deployment_failed",
|
|
"validation": {"validation_gate_version": 1, "required_contexts": ["Managed validation / full (push)"],
|
|
"timeout_seconds": 1800, "poll_interval_seconds": 5, "max_poll_interval_seconds": 30,
|
|
"last_result": "pass", "observed": [{"context": "Managed validation / full (push)", "status": "success"}]},
|
|
"audit": [{"stage": "validation_passed"}, {"stage": "deployment_started"},
|
|
{"stage": "deployment_failed", "reason": "deployment_failed"}]}
|
|
inbox = m.STATE_ROOT / "webhook-inbox"
|
|
inbox.mkdir(mode=0o700)
|
|
marker = inbox / f"{semantic}.json"
|
|
marker.write_text(json.dumps(record))
|
|
marker.chmod(0o600)
|
|
return marker, record
|
|
|
|
|
|
@pytest.mark.parametrize("damage", [None, "deployed", "pending", "missing", "metadata", "ancestry"])
|
|
def test_prior_main_attempt_requires_private_green_failure_and_git_proof(bridge, monkeypatch, damage):
|
|
m, _project, _repo, _revision, metadata, path, events = bridge
|
|
prior_revision = "b" * 40
|
|
state = json.loads(path.read_bytes())
|
|
prior_metadata = {**metadata, "revision": prior_revision, "id": prior_revision[:12]}
|
|
state["last_attempt"].update(revision=prior_revision, metadata=prior_metadata)
|
|
path.write_text(json.dumps(state))
|
|
marker, record = prior_record(m, prior_revision)
|
|
monkeypatch.setattr(m, "deployment_metadata", lambda *_a, **_k: prior_metadata if damage != "metadata" else {**prior_metadata, "tree": "e" * 40})
|
|
def ancestry(command, **_kwargs):
|
|
assert command[-4:] == ["merge-base", "--is-ancestor", prior_revision, metadata["revision"]]
|
|
if damage == "ancestry":
|
|
raise subprocess.CalledProcessError(1, command)
|
|
monkeypatch.setattr(m.subprocess, "run", ancestry)
|
|
if damage == "deployed":
|
|
record["status"] = "deployed" # Retain the old failure audit deliberately.
|
|
elif damage == "pending":
|
|
record["validation"]["observed"][0]["status"] = "pending"
|
|
marker.write_text(json.dumps(record))
|
|
if damage == "missing":
|
|
marker.unlink()
|
|
if damage is not None:
|
|
with pytest.raises(m.AppOpsLegacyNoMutation):
|
|
prepare_bridge(bridge)
|
|
assert events == []
|
|
else:
|
|
migration = prepare_bridge(bridge)
|
|
assert migration["intent"]["migrationProof"]["priorAttempt"]["metadataSha256"] == m._canonical_json_sha256(prior_metadata)
|
|
assert events == ["backup", "snapshot", "tag"]
|