7173956e12
The facility stayed at one replica and the relay credential was not read. Offsite publication still waits for a prefix-scoped publisher grant. Assistant: grok Assistant-Session: 01a198fa-fd27-7e41-a0a2-57355c3e4efd
80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""MAS evidence capture and rehearsal. No cluster and no provider."""
|
|
|
|
import importlib.util
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def load(name):
|
|
path = Path(__file__).resolve().parents[1] / "scripts" / name
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
archive = load("mas_evidence_archive.py")
|
|
primary = load("primary_application_archive.py")
|
|
|
|
|
|
def tree(tmp_path):
|
|
source = tmp_path / "evidence"
|
|
source.mkdir()
|
|
(source / "events.jsonl").write_text('{"kind":"accepted"}\n')
|
|
for name in ("tickets.sqlite3", "tickets-outbox.sqlite3"):
|
|
connection = sqlite3.connect(source / name)
|
|
connection.execute("CREATE TABLE item (id INTEGER PRIMARY KEY, note TEXT)")
|
|
connection.execute("INSERT INTO item (note) VALUES ('held')")
|
|
connection.commit()
|
|
connection.close()
|
|
return source
|
|
|
|
|
|
def test_profile_is_separate_from_email_connect():
|
|
key = primary.key_for("mas", "2026-11-02T01:20:00Z")
|
|
assert key.startswith("platform-pg/application-archives/mas/")
|
|
with pytest.raises(ValueError):
|
|
primary.validate_key("email-connect", key)
|
|
|
|
|
|
def test_capture_refuses_unless_writers_stay_at_one_replica(tmp_path):
|
|
with pytest.raises(archive.Refused, match="writers_not_quiesced"):
|
|
archive.capture(
|
|
tree(tmp_path),
|
|
tmp_path / "out",
|
|
captured_at="2026-11-02T01:20:00Z",
|
|
image="mas-live:test",
|
|
writers_quiesced=False,
|
|
replicas={"mas-mx": 1, "mas-statute": 1},
|
|
)
|
|
|
|
|
|
def test_seal_and_rehearse_do_not_restore_or_send(tmp_path):
|
|
source = tree(tmp_path)
|
|
captured = archive.capture(
|
|
source,
|
|
tmp_path / "captured",
|
|
captured_at="2026-11-02T01:20:00Z",
|
|
image="mas-live:test",
|
|
writers_quiesced=True,
|
|
replicas={"mas-mx": 1, "mas-statute": 1},
|
|
)
|
|
identity = tmp_path / "identity.txt"
|
|
__import__("subprocess").check_call(["age-keygen", "-o", str(identity)])
|
|
recipient = __import__("subprocess").check_output(
|
|
["age-keygen", "-y", str(identity)], text=True
|
|
).strip()
|
|
sealed = archive.seal(tmp_path / "captured", tmp_path / "stage", recipient)
|
|
proof = archive.rehearse(tmp_path / "stage" / "packet.tar.age", identity, tmp_path / "rehearsal")
|
|
assert captured["mail_sent"] is False
|
|
assert proof == {
|
|
"status": "rehearsal_matched",
|
|
"application": "mas",
|
|
"ciphertext_sha256": sealed["ciphertext_sha256"],
|
|
"mail_sent": False,
|
|
"production_restored": False,
|
|
"facility_scaled": False,
|
|
}
|
|
assert [path.name for path in (tmp_path / "rehearsal").iterdir()] == ["rehearsal.json"]
|