"""Supervisor lifecycle against a synthetic Kubernetes/backup adapter, no network.""" import copy import importlib.util import json import os import stat from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("source_date_supervisor", ROOT / "scripts/source_date_supervisor.py") supervisor = importlib.util.module_from_spec(spec) spec.loader.exec_module(supervisor) IMAGE = "forgejo.coulomb.social/coulomb/coulomb-social@sha256:" + "a" * 64 BASE = {"tables_sha256": "b" * 64, "files_sha256": "c" * 64, "tables": 13, "rows": 112, "files": 82} AFTER = {**BASE, "tables_sha256": "d" * 64, "files_sha256": "e" * 64} class FakeNode(supervisor.Node): """Only control-plane/capture effects are simulated; production paths never open.""" def __init__(self): self.dep = {"metadata": {"uid": "deployment-id", "resourceVersion": "17"}, "spec": { "replicas": 1, "strategy": {"type": "Recreate"}, "template": {"spec": { "containers": [{"name": "app", "image": IMAGE, "env": [{"name": "CONTENT_ROOT", "value": "/app/var/content"}], "envFrom": [{"secretRef": {"name": "synthetic-app-env"}}], "volumeMounts": [{"name": "content", "mountPath": "/app/var/content"}]}], "volumes": [{"name": "content", "persistentVolumeClaim": { "claimName": supervisor.CLAIM}}]}}}} self.fp, self.helper, self.journal = BASE, None, "absent" self.calls, self.fail, self.capture_count, self.apply_count = [], None, 0, 0 self.next_uid = 1 self.busy = False self.sidecar = False self.extra_writer = False self.running_image = None def trip(self, name): if self.fail == name: self.fail = None raise supervisor.Refused("synthetic lost response: " + name) def doc(self, *args): kind = args[1] if kind == "namespace": return {"metadata": {"uid": "cluster-id"}} if kind == "node": return {"metadata": {"uid": "node-id", "name": "synthetic-node"}, "spec": {}, "status": { "conditions": [{"type": "Ready", "status": "True"}]}} if kind == "pvc": return {"metadata": {"uid": "pvc-id"}, "status": {"phase": "Bound"}} if kind in {"hpa", "applications.argoproj.io"}: return {"items": []} if kind == "deployment": return copy.deepcopy(self.dep) if kind == "replicaset": return {"metadata": {"ownerReferences": [{"uid": "deployment-id"}]}} if kind == "pods": pods = [] if self.dep["spec"]["replicas"]: pods.append({"metadata": {"name": "synthetic-app", "ownerReferences": [ {"kind": "ReplicaSet", "name": "synthetic-rs"}]}, "spec": { "nodeName": "synthetic-node", "containers": [{"image": self.running_image or self.dep["spec"]["template"]["spec"]["containers"][0]["image"]}], "volumes": [{"persistentVolumeClaim": { "claimName": supervisor.CLAIM}}]}, "status": { "phase": "Running", "conditions": [ {"type": "Ready", "status": "True"}], "containerStatuses": [{"imageID": self.dep["spec"]["template"][ "spec"]["containers"][0]["image"]}]}}) if self.helper: pods.append(copy.deepcopy(self.helper)) if self.extra_writer: pods.append({"metadata": {"name": "unreviewed-writer"}, "spec": {"volumes": [ {"persistentVolumeClaim": {"claimName": supervisor.CLAIM}}]}, "status": {"phase": "Running"}}) return {"items": pods} raise AssertionError(args) def command(self, args, payload=None, **kwargs): self.calls.append(args[:4]) if args[:2] == ["helm", "history"]: return supervisor.canonical([{"revision": 18, "status": "deployed"}]) if args[:3] == ["helm", "get", "manifest"]: return b"synthetic helm manifest" if args[:2] == ["kubectl", "exec"]: message = json.loads(payload) mode = message["mode"] if mode == "prepare": return supervisor.canonical({"fingerprint": self.fp, "storage_constraint_verified": True, "expected_after": AFTER}) if mode == "probe": return supervisor.canonical({"fingerprint": self.fp}) if mode == "inspect": status = "operation_busy" if self.busy else self.journal return supervisor.canonical({"status": status, "fingerprint": self.fp}) if mode == "reconcile": assert self.journal == "pending" self.journal, self.fp = "rolled_back", BASE return supervisor.canonical({"status": self.journal, "fingerprint": self.fp}) assert mode == "apply" self.apply_count += 1 if self.fail == "apply_pending": self.journal = "pending" self.trip("apply_pending") self.journal, self.fp = "completed", AFTER self.trip("apply_reply") return supervisor.canonical({"status": "completed", "whole_store_postflight": True, "fingerprint": self.fp}) if args[:2] == ["kubectl", "scale"]: current = next(a for a in args if a.startswith("--current-replicas=")).split("=")[1] assert int(current) == self.dep["spec"]["replicas"] wanted = next(a for a in args if a.startswith("--replicas=")).split("=")[1] self.dep["spec"]["replicas"] = int(wanted) return b"" if args[:2] == ["kubectl", "create"]: pod = json.loads(payload) if self.sidecar: pod["spec"]["containers"].append({"name": "injected"}) if "--dry-run=server" not in args: assert self.dep["spec"]["replicas"] == 0 and not self.helper pod["metadata"]["uid"] = "helper-" + str(self.next_uid) self.next_uid += 1 pod["status"] = {"phase": "Running"} self.helper = copy.deepcopy(pod) self.trip("create_reply") return supervisor.canonical(pod) if args[:3] == ["kubectl", "delete", "--raw"]: assert json.loads(payload)["preconditions"]["uid"] == self.helper["metadata"]["uid"] self.helper = None self.trip("delete_reply") return b"" if args[:3] == ["kubectl", "rollout", "status"]: self.trip("resume_reply") return b"" assert args[:2] == ["kubectl", "wait"], args return b"" def verify_backup_installation(self, target): self.trip("helper_hash") def verify_local_node(self, node): pass # synthetic cluster/root adapter, never a native machine-id claim def installation_pins(self): return {"installed_helpers": {name: "f" * 64 for name in supervisor.HELPERS}, "public_recipient": "age1synthetic"} def capture(self, target): self.capture_count += 1 self.trip("capture") return {"status": "synthetic-capture"} def lock_backups(self, stack): self.trip("backup_lock") def helper_directory(self, path): path.mkdir(mode=0o700, exist_ok=True) def read_state(self, path): return json.loads(path.read_bytes()) @pytest.fixture def prepared(tmp_path): node = FakeNode() target = {"cluster_uid": "cluster-id", "deployment_uid": "deployment-id", "node_uid": "node-id", "node_name": "synthetic-node", "pvc_uid": "pvc-id", "image": IMAGE, "deployment_spec_sha256": supervisor.sha(supervisor.canonical(node.dep["spec"])), "helm_revision": 18, "helm_manifest_sha256": supervisor.sha(b"synthetic helm manifest"), "public_recipient": "age1synthetic", "installed_helpers": { name: "f" * 64 for name in supervisor.HELPERS}} capsule = tmp_path / "capsule" capsule.mkdir(mode=0o700) for name in supervisor.MEMBERS: (capsule / name).write_text("synthetic source") (capsule / "plan.json").write_text(json.dumps({ "schema": "coulomb.source-date-backfill-plan.v1", "blockers": [], "prepared_change_set_valid": True})) manifest = {"schema": "coulomb.supervised-date-capsule.v1", "baseline": BASE, "target": target, "attempt_id": "1" * 32, "requires_founder_attended_execution": True, "files": {name: supervisor.sha((capsule / name).read_bytes()) for name in supervisor.MEMBERS}} (capsule / "manifest.json").write_bytes(supervisor.canonical(manifest)) digest = "sha256:" + supervisor.sha(supervisor.canonical(manifest)) frozen = supervisor.Capsule(capsule, digest) root = tmp_path / "state" return supervisor.Supervisor(frozen, node, state_root=root), node, capsule, digest def test_success_captures_fences_applies_checks_and_resumes_once(prepared): operation, node, _, _ = prepared assert not operation.state_root.exists() operation.preflight() assert not operation.state_root.exists() and node.capture_count == 0 result = operation.execute() assert result["phase"] == "completed" and result["correction_status"] == "completed" assert node.capture_count == node.apply_count == 1 assert node.dep["spec"]["replicas"] == 1 and not node.helper assert result["helper_evidence_root_custody_verified"] with pytest.raises(supervisor.Refused, match="prior_attempt"): operation.execute() assert node.apply_count == 1 @pytest.mark.parametrize("failure", ["apply_reply", "apply_pending", "create_reply", "delete_reply"]) def test_uncertainty_preserves_fence_and_recovers_without_new_capture_or_apply(prepared, failure): operation, node, _, _ = prepared node.fail = failure with pytest.raises(supervisor.Refused): operation.execute() assert node.dep["spec"]["replicas"] == 0 applied = node.apply_count result = operation.recover() assert result["phase"] == "completed" and node.dep["spec"]["replicas"] == 1 assert node.capture_count == 1 and node.apply_count == applied assert not node.helper if failure == "apply_pending": assert result["correction_status"] == "rolled_back" def test_live_original_operation_blocks_recovery_and_never_restarts(prepared): operation, node, _, _ = prepared node.fail = "apply_pending" with pytest.raises(supervisor.Refused): operation.execute() node.busy = True with pytest.raises(supervisor.Refused, match="still_running"): operation.recover() assert node.capture_count == node.apply_count == 1 and node.dep["spec"]["replicas"] == 0 def test_lost_resume_reply_preserves_new_productive_writes(prepared): operation, node, _, _ = prepared node.fail = "resume_reply" with pytest.raises(supervisor.Refused): operation.execute() assert node.dep["spec"]["replicas"] == 1 node.fp = {**AFTER, "rows": 113} # legitimate write after the application resumed result = operation.recover() assert result["post_resume_writes_preserved"] and node.fp["rows"] == 113 assert node.capture_count == node.apply_count == 1 @pytest.mark.parametrize("problem", ["pod_injection", "foreign_writer", "spec_drift", "helper_hash", "pod_image"]) def test_preflight_refusals_never_capture_or_scale(prepared, problem): operation, node, _, _ = prepared if problem == "pod_injection": node.sidecar = True elif problem == "foreign_writer": node.extra_writer = True elif problem == "spec_drift": node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0" elif problem == "pod_image": node.running_image = IMAGE + "0" else: node.fail = "helper_hash" with pytest.raises(supervisor.Refused): operation.execute() assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1 def test_changed_capsule_refuses_before_any_backend_access(prepared): _, _, path, digest = prepared (path / "wire.py").write_text("changed") with pytest.raises(supervisor.Refused, match="member_changed"): supervisor.Capsule(path, digest) def test_unrelated_data_drift_after_uncertain_result_never_resumes(prepared): operation, node, _, _ = prepared node.fail = "apply_reply" with pytest.raises(supervisor.Refused): operation.execute() node.fp = {**AFTER, "rows": 999} with pytest.raises(supervisor.Refused, match="dataset_requires_owner_review"): operation.recover() assert node.dep["spec"]["replicas"] == 0 and node.helper assert node.capture_count == node.apply_count == 1 def test_replaced_helper_is_never_deleted_or_used_for_recovery(prepared): operation, node, _, _ = prepared node.fail = "apply_reply" with pytest.raises(supervisor.Refused): operation.execute() node.helper["metadata"]["uid"] = "foreign-helper" with pytest.raises(supervisor.Refused, match="helper_replaced"): operation.recover() assert node.dep["spec"]["replicas"] == 0 assert node.helper["metadata"]["uid"] == "foreign-helper" def test_terminal_original_helper_is_uid_deleted_and_recreated_only_for_reconciliation(prepared): operation, node, _, _ = prepared node.fail = "apply_reply" with pytest.raises(supervisor.Refused): operation.execute() node.helper["status"]["phase"] = "Succeeded" result = operation.recover() assert result["correction_status"] == "completed" and not node.helper assert node.next_uid == 3 and node.capture_count == node.apply_count == 1 def test_capsule_preparation_is_private_fresh_and_has_a_new_attempt_id(prepared, tmp_path): import runpy import sys operation, _, source, _ = prepared sys.path.insert(0, str(ROOT / "scripts")) try: prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"] finally: sys.path.pop(0) baseline, target = tmp_path / "baseline.json", tmp_path / "target.json" baseline.write_bytes(supervisor.canonical(BASE)) target.write_bytes(supervisor.canonical(operation.target)) proofs = [] for name in ("first", "second"): output = tmp_path / name proof = prepare(source / "plan.json", baseline, target, output) capsule = supervisor.Capsule(output, proof["proposal_digest"]) assert capsule.manifest["requires_founder_attended_execution"] assert output.stat().st_mode & 0o777 == 0o700 assert all(p.stat().st_mode & 0o777 == 0o600 for p in output.iterdir()) proofs.append(proof["proposal_digest"]) with pytest.raises(ValueError, match="Fresh private"): prepare(source / "plan.json", baseline, target, output) assert proofs[0] != proofs[1] @pytest.fixture def native_observer_tmp(): # The native observer deliberately requires /tmp, regardless of TMPDIR. import tempfile with tempfile.TemporaryDirectory(prefix="csoc-observer-test-", dir="/tmp") as directory: yield Path(directory) @pytest.mark.parametrize("changed_image", [False, True]) def test_observer_prepares_without_capturing_or_refuses_missing_release( prepared, native_observer_tmp, monkeypatch, changed_image): import runpy import sys operation, node, source, _ = prepared sys.path.insert(0, str(ROOT / "scripts")) try: observer = runpy.run_path(str(ROOT / "scripts/observe-source-date-target.py"))["observe"] finally: sys.path.pop(0) monkeypatch.setattr(supervisor.os, "geteuid", lambda: 0) # simulated root context only output = native_observer_tmp / "observed-capsule" if changed_image: node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0" with pytest.raises(RuntimeError, match="prerequisite_missing"): observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id", deployment_uid="deployment-id", pvc_uid="pvc-id", node=node) assert not output.exists() else: proof = observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id", deployment_uid="deployment-id", pvc_uid="pvc-id", node=node) capsule = supervisor.Capsule(output, proof["proposal_digest"]) assert capsule.baseline == BASE and capsule.target == operation.target assert proof["baseline_rechecked"] and not proof["live_changes"] assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1 def test_closeout_failure_recovers_only_custody_and_preserves_new_writes(prepared, monkeypatch): operation, node, _, _ = prepared seal = node.seal_helper_evidence def lost_closeout(path, expected): assert node.dep['spec']['replicas'] == 1 and node.helper is None seal(path, expected) raise supervisor.Refused('synthetic lost custody response') monkeypatch.setattr(node, 'seal_helper_evidence', lost_closeout) with pytest.raises(supervisor.Refused, match='lost custody'): operation.execute() assert operation.state['failed_phase'] == 'custody_closeout_started' node.fp = {**AFTER, 'rows': 115} before = list(node.calls) monkeypatch.setattr(node, 'seal_helper_evidence', seal) result = operation.recover() assert result['phase'] == 'completed' and result['helper_evidence_root_custody_verified'] assert node.fp['rows'] == 115 and node.capture_count == node.apply_count == 1 assert all(call[:2] not in (['kubectl', 'create'], ['kubectl', 'scale'], ['kubectl', 'exec']) for call in node.calls[len(before):]) with pytest.raises(supervisor.Refused, match='completed_attempt'): operation.recover() def test_uncertain_helper_evidence_is_not_sealed_before_resume(prepared, monkeypatch): from unittest.mock import Mock operation, node, _, _ = prepared seal = Mock() monkeypatch.setattr(node, 'seal_helper_evidence', seal) node.fail = 'apply_pending' with pytest.raises(supervisor.Refused): operation.execute() seal.assert_not_called() assert node.helper and node.dep['spec']['replicas'] == 0 def evidence_directory(tmp_path): private = tmp_path / 'private' private.mkdir(mode=0o700) for name, data in {'execute.lock': b'', 'journal.json': b'synthetic private journal', 'result.json': b'synthetic result'}.items(): file = private / name file.write_bytes(data) file.chmod(0o600) return private def test_evidence_sealing_preserves_bytes_paths_and_modes(tmp_path): private = evidence_directory(tmp_path) node = supervisor.Node() before = node.helper_evidence(private) node.seal_helper_evidence(private, before) assert node.helper_evidence(private) == before for path in [private, *private.iterdir()]: info = path.stat() assert (info.st_uid, info.st_gid) == (os.geteuid(), os.getegid()) assert stat.S_IMODE(info.st_mode) == (0o700 if path.is_dir() else 0o600) @pytest.mark.parametrize('problem', ['changed_bytes', 'extra', 'symlink', 'hardlink', 'broad', 'busy', 'fifo']) def test_evidence_closeout_refuses_drift_before_any_chown(tmp_path, monkeypatch, problem): from unittest.mock import Mock private = evidence_directory(tmp_path) node = supervisor.Node() before = node.helper_evidence(private) item = private / 'result.json' lock = None if problem == 'changed_bytes': item.write_bytes(b'changed') elif problem == 'extra': (private / 'unexpected.json').write_bytes(b'unexpected') elif problem in {'symlink', 'hardlink', 'fifo'}: item.unlink() if problem == 'symlink': item.symlink_to(private / 'journal.json') elif problem == 'hardlink': os.link(private / 'journal.json', item) else: os.mkfifo(item, 0o600) elif problem == 'broad': item.chmod(0o644) else: lock = (private / 'execute.lock').open() supervisor.fcntl.flock(lock, supervisor.fcntl.LOCK_EX | supervisor.fcntl.LOCK_NB) chown = Mock() monkeypatch.setattr(supervisor.os, 'fchown', chown) try: with pytest.raises((supervisor.Refused, OSError)): node.seal_helper_evidence(private, before) chown.assert_not_called() finally: if lock: lock.close() @pytest.mark.parametrize('overlap', ['helper', 'ancestor', 'child']) def test_closeout_refuses_existing_helper_or_host_mount(prepared, monkeypatch, overlap): from unittest.mock import Mock operation, node, _, _ = prepared operation.execute() original = node.doc evidence_path = operation.directory / 'private' pod = {'metadata': {'name': operation.helper if overlap == 'helper' else 'foreign'}, 'spec': {'volumes': []}} if overlap != 'helper': path = evidence_path.parent if overlap == 'ancestor' else evidence_path / 'journal.json' pod['spec']['volumes'] = [{'hostPath': {'path': str(path)}}] def documents(*args): result = original(*args) if args[:2] == ('get', 'pods'): result['items'].append(pod) return result monkeypatch.setattr(node, 'doc', documents) seal = Mock() monkeypatch.setattr(node, 'seal_helper_evidence', seal) with pytest.raises(supervisor.Refused, match='still_'): operation.complete() seal.assert_not_called() def test_partial_helper_ownership_transfer_resumes_from_fingerprints(tmp_path, monkeypatch): from types import SimpleNamespace private = evidence_directory(tmp_path) node = supervisor.Node() real_stat = os.fstat owners = {p.stat().st_ino: (10001, 10001) for p in [private, *private.iterdir()]} changed = [] def observed(fd): info = real_stat(fd) values = {name: getattr(info, name) for name in dir(info) if name.startswith('st_')} values['st_uid'], values['st_gid'] = owners.get(info.st_ino, (info.st_uid, info.st_gid)) return SimpleNamespace(**values) def transfer(fd, uid, gid): inode = real_stat(fd).st_ino owners[inode] = (uid, gid) changed.append(inode) if len(changed) == 1: raise OSError('synthetic interruption after first ownership change') monkeypatch.setattr(supervisor.os, 'fstat', observed) monkeypatch.setattr(supervisor.os, 'fchown', transfer) expected = node.helper_evidence(private) with pytest.raises(OSError, match='interruption'): node.seal_helper_evidence(private, expected) assert len(set(owners.values())) == 2 node.seal_helper_evidence(private, expected) assert set(owners.values()) == {(os.geteuid(), os.getegid())} assert node.helper_evidence(private) == expected def test_second_lost_reply_during_recovery_uses_latest_phase(prepared): operation, node, _, _ = prepared node.fail = 'apply_reply' with pytest.raises(supervisor.Refused): operation.execute() node.fail = 'resume_reply' with pytest.raises(supervisor.Refused): operation.recover() assert node.dep['spec']['replicas'] == 1 assert operation.state['phase'] == 'resume_started' node.fp = {**AFTER, 'rows': 116} result = operation.recover() assert result['phase'] == 'completed' and result['post_resume_writes_preserved'] assert node.fp['rows'] == 116 and node.capture_count == node.apply_count == 1