tegwick
d9da608e65
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a192e9-c28b-7533-bf25-e2341364c22a
555 lines
24 KiB
Python
555 lines
24 KiB
Python
"""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
|