tegwick
904dcb3ce3
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
238 lines
11 KiB
Python
238 lines
11 KiB
Python
"""Synthetic release safety cases; no cluster, credentials or real content."""
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1] / "tools/coulomb_image_release_node.py"
|
|
spec = importlib.util.spec_from_file_location("storage_release", SOURCE)
|
|
release = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(release)
|
|
|
|
launcher_spec = importlib.util.spec_from_file_location("storage_launcher", SOURCE.with_name("coulomb-image-release.py"))
|
|
launcher = importlib.util.module_from_spec(launcher_spec)
|
|
launcher_spec.loader.exec_module(launcher)
|
|
|
|
|
|
def capsule(tmp_path):
|
|
root = tmp_path / "capsule"
|
|
root.mkdir()
|
|
(root / "probe.py").write_bytes(b"synthetic read-only probe")
|
|
plan = {"files": {"probe.py": release.sha((root / "probe.py").read_bytes())}}
|
|
(root / "manifest.json").write_text(json.dumps(plan))
|
|
return root, "sha256:" + release.sha(release.canonical(plan))
|
|
|
|
|
|
def test_capsule_tampering_refuses_before_privileged_work(tmp_path, monkeypatch):
|
|
root, digest = capsule(tmp_path)
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
|
|
(root / "probe.py").write_bytes(b"changed")
|
|
with pytest.raises(release.Refused, match="capsule_file_differs"):
|
|
release.Release(root, digest)
|
|
assert not (tmp_path / "journal").exists()
|
|
|
|
|
|
def test_preflight_never_creates_an_execution_journal(tmp_path, monkeypatch):
|
|
root, digest = capsule(tmp_path)
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
|
|
instance = release.Release(root, digest, read_only=True)
|
|
instance.save()
|
|
assert not (tmp_path / "journal").exists()
|
|
|
|
|
|
def test_prior_attempt_is_not_blindly_replayed(tmp_path, monkeypatch):
|
|
root, digest = capsule(tmp_path)
|
|
journal = tmp_path / "journal"
|
|
(journal / digest.removeprefix("sha256:")).mkdir(parents=True)
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", journal)
|
|
with pytest.raises(release.Refused, match="prior_execution_requires_review"):
|
|
release.Release(root, digest)
|
|
|
|
|
|
def test_mutable_staging_is_frozen_before_live_work(tmp_path, monkeypatch):
|
|
root, digest = capsule(tmp_path)
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
|
|
instance = release.Release(root, digest)
|
|
(root / "probe.py").write_bytes(b"changed after verification")
|
|
assert (instance.root / "probe.py").read_bytes() == b"synthetic read-only probe"
|
|
assert (instance.journal / "receipt.json").is_file()
|
|
|
|
|
|
def test_nested_capsule_and_release_root_stay_private_under_normal_umask(tmp_path, monkeypatch):
|
|
root, _ = capsule(tmp_path)
|
|
nested = root / "charts/coulomb/templates/deployment.yaml"
|
|
nested.parent.mkdir(parents=True)
|
|
nested.write_bytes(b"synthetic chart")
|
|
plan = {"files": {str(p.relative_to(root)): release.sha(p.read_bytes())
|
|
for p in root.rglob("*") if p.is_file() and p.name != "manifest.json"}}
|
|
(root / "manifest.json").write_text(json.dumps(plan))
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
|
|
previous = os.umask(0o022)
|
|
try:
|
|
instance = release.Release(root, "sha256:" + release.sha(release.canonical(plan)))
|
|
finally:
|
|
os.umask(previous)
|
|
for path in [tmp_path / "journal", *(tmp_path / "journal").rglob("*")]:
|
|
assert stat.S_IMODE(path.stat().st_mode) == (0o700 if path.is_dir() else 0o600)
|
|
assert (instance.root / nested.relative_to(root)).read_bytes() == b"synthetic chart"
|
|
|
|
|
|
@pytest.mark.parametrize("problem", ["broad", "link", "ancestor_link", "file"])
|
|
def test_unsafe_release_parent_is_refused_without_changing_it(tmp_path, monkeypatch, problem):
|
|
root, digest = capsule(tmp_path)
|
|
journal = tmp_path / "journal"
|
|
if problem == "broad":
|
|
journal.mkdir(mode=0o755)
|
|
elif problem == "file":
|
|
journal.write_text("untouched")
|
|
elif problem == "link":
|
|
journal.symlink_to(root, target_is_directory=True)
|
|
else:
|
|
alias = tmp_path / "alias"
|
|
alias.symlink_to(root, target_is_directory=True)
|
|
journal = alias / "journal"
|
|
monkeypatch.setattr(release.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(release, "JOURNAL_ROOT", journal)
|
|
before = {str(p): p.lstat() for p in tmp_path.rglob("*")}
|
|
with pytest.raises((release.Refused, OSError)):
|
|
release.Release(root, digest)
|
|
assert before == {str(p): p.lstat() for p in tmp_path.rglob("*")}
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["business_rows_sha256", "content_files_sha256", "migration_rows_sha256"])
|
|
def test_dataset_drift_is_not_accepted(field):
|
|
instance = object.__new__(release.Release)
|
|
baseline = {"business_tables": 2, "business_rows": 3, "business_rows_sha256": "synthetic-a",
|
|
"content_files": 4, "content_files_sha256": "synthetic-b",
|
|
"migration_rows_sha256": "synthetic-migrations"}
|
|
instance.plan = {"expected_data": baseline}
|
|
changed = dict(baseline, duplicate_slug_groups=0, tenant_slug_constraint=True)
|
|
changed[field] = "different"
|
|
with pytest.raises(release.Refused, match="dataset_drifted"):
|
|
instance.same_data(changed)
|
|
|
|
|
|
def test_failed_preflight_never_captures_or_upgrades():
|
|
instance = object.__new__(release.Release)
|
|
instance.preflight = Mock(side_effect=release.Refused("guard stopped"))
|
|
instance.capture = Mock()
|
|
instance.command = Mock()
|
|
with pytest.raises(release.Refused):
|
|
instance.execute()
|
|
instance.capture.assert_not_called()
|
|
instance.command.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("rollback_spec,rollback_ready,verified", [
|
|
({"synthetic": "old"}, 1, True),
|
|
({"synthetic": "candidate"}, 1, False),
|
|
({"synthetic": "old"}, 0, False),
|
|
])
|
|
def test_postflight_failure_rolls_back_image_without_restoring_data(
|
|
tmp_path, monkeypatch, rollback_spec, rollback_ready, verified):
|
|
pin = tmp_path / "snapshot.py"
|
|
pin.write_bytes(b"old pin")
|
|
monkeypatch.setattr(release, "SNAPSHOT", pin)
|
|
instance = object.__new__(release.Release)
|
|
instance.receipt = {}
|
|
instance.root = tmp_path
|
|
spec = {"synthetic": "candidate"}
|
|
baseline = {"business_tables": 2, "business_rows": 3, "business_rows_sha256": "synthetic-a",
|
|
"content_files": 4, "content_files_sha256": "synthetic-b",
|
|
"migration_rows_sha256": "synthetic-migrations"}
|
|
instance.plan = {"candidate_spec_sha256": release.sha(release.canonical(spec)), "baseline_revision": 18,
|
|
"baseline_spec_sha256": release.sha(release.canonical({"synthetic": "old"})),
|
|
"expected_data": baseline}
|
|
instance.preflight = Mock()
|
|
instance.capture = Mock()
|
|
instance.save = Mock()
|
|
instance.deployment = Mock(side_effect=[
|
|
{"spec": copy.deepcopy(spec), "status": {"readyReplicas": 1}},
|
|
{"spec": rollback_spec, "status": {"readyReplicas": rollback_ready}},
|
|
])
|
|
instance.probe = lambda mode: dict(baseline, business_rows_sha256="changed")
|
|
instance.command = Mock(return_value=SimpleNamespace(stdout=b"", returncode=0))
|
|
instance.restore_pin = Mock()
|
|
with pytest.raises(release.Refused, match="dataset_drifted"):
|
|
instance.execute()
|
|
commands = [call.args[0] for call in instance.command.call_args_list]
|
|
assert commands[0][:2] == ["helm", "upgrade"] and "--atomic" in commands[0]
|
|
assert commands[1][:4] == ["helm", "rollback", "coulomb-social", "18"]
|
|
assert not any("pg_restore" in command for command in commands)
|
|
instance.restore_pin.assert_called_once_with(b"old pin")
|
|
assert instance.receipt["image_rollback_completed"] is verified
|
|
|
|
|
|
def test_live_launcher_rejects_unattended_execution_before_ssh(monkeypatch):
|
|
_, digest, _ = launcher.prepare()
|
|
monkeypatch.setattr(launcher.sys, "argv", ["release", "--execute", "--expect-digest", digest])
|
|
monkeypatch.setattr(launcher.sys, "stdin", SimpleNamespace(isatty=lambda: False))
|
|
command = Mock()
|
|
monkeypatch.setattr(launcher.subprocess, "run", command)
|
|
with pytest.raises(SystemExit, match="own terminal"):
|
|
launcher.main()
|
|
command.assert_not_called()
|
|
|
|
|
|
def test_timeout_gives_the_existing_capture_resume_window(monkeypatch):
|
|
instance = object.__new__(release.Release)
|
|
instance.env = {}
|
|
process = Mock()
|
|
process.communicate.side_effect = release.subprocess.TimeoutExpired(["synthetic"], 1)
|
|
monkeypatch.setattr(release.subprocess, "Popen", Mock(return_value=process))
|
|
with pytest.raises(release.Refused, match="cleanup_window"):
|
|
instance.command(["synthetic"], timeout=1)
|
|
process.terminate.assert_called_once()
|
|
process.wait.assert_called_once_with(timeout=450)
|
|
process.kill.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["global_slug_constraint", "migration_0003_applied"])
|
|
@pytest.mark.parametrize("value", [False, None])
|
|
def test_image_only_release_requires_existing_migration_and_constraint(field, value):
|
|
instance = object.__new__(release.Release)
|
|
baseline = {"business_tables": 2, "business_rows": 3,
|
|
"business_rows_sha256": "rows", "content_files": 4,
|
|
"content_files_sha256": "files", "migration_rows_sha256": "migrations"}
|
|
instance.plan = {"expected_data": baseline}
|
|
observed = dict(baseline, duplicate_slug_groups=0, tenant_slug_constraint=True,
|
|
global_slug_constraint=True, migration_0003_applied=True)
|
|
instance.same_data(observed)
|
|
observed[field] = value
|
|
with pytest.raises(release.Refused, match="already_migrated_baseline_required"):
|
|
instance.same_data(observed)
|
|
|
|
|
|
def test_rollback_does_not_overwrite_unrecognized_snapshot_helper(tmp_path, monkeypatch):
|
|
pin = tmp_path / "snapshot.py"
|
|
pin.write_bytes(b"concurrent owner change")
|
|
monkeypatch.setattr(release, "SNAPSHOT", pin)
|
|
instance = object.__new__(release.Release)
|
|
instance.plan = {"files": {"release/identity-export-snapshot.py": release.sha(b"new pin")}}
|
|
with pytest.raises(release.Refused, match="snapshot_pin_changed_during_rollback"):
|
|
instance.restore_pin(b"old pin")
|
|
assert pin.read_bytes() == b"concurrent owner change"
|
|
|
|
|
|
def test_known_candidate_helper_can_be_rolled_back(tmp_path, monkeypatch):
|
|
pin = tmp_path / "snapshot.py"
|
|
pin.write_bytes(b"new pin")
|
|
monkeypatch.setattr(release, "SNAPSHOT", pin)
|
|
instance = object.__new__(release.Release)
|
|
instance.plan = {"files": {"release/identity-export-snapshot.py": release.sha(b"new pin")}}
|
|
instance.restore_pin(b"old pin")
|
|
assert pin.read_bytes() == b"old pin"
|
|
# A completed inverse can be checked again without changing it.
|
|
instance.restore_pin(b"old pin")
|
|
assert pin.read_bytes() == b"old pin"
|