edd8687ea6
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 4541@bnt-lap001 Assistant-Session: 2099b3aa-d7d9-49a9-ab4e-fa8b7d0e39e8
74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
"""MAS backup failure notices: queued privately, delivered with a verified echo, replayed."""
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("notify", ROOT / "scripts/mas_evidence_notify.py")
|
|
nt = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(nt)
|
|
|
|
|
|
class Hub:
|
|
def __init__(self, down=False):
|
|
self.down, self.sent = down, []
|
|
|
|
def __call__(self, base, message):
|
|
if self.down:
|
|
raise OSError("hub unreachable")
|
|
self.sent.append(message)
|
|
return f"00000000-0000-0000-0000-{len(self.sent):012d}"
|
|
|
|
|
|
def test_backup_failure_is_delivered_to_platform_root(tmp_path):
|
|
hub = Hub()
|
|
result = nt.notify(tmp_path / "n", unit="mas-evidence-backup.service", sender=hub)
|
|
assert result["status"] == "inbox_delivery_verified" and len(result["message_ids"]) == 1
|
|
message = hub.sent[0]
|
|
assert message["to_agent"] == "platform-root" and message["subject"] == "MAS evidence backup failure"
|
|
body = json.loads(message["body"])
|
|
assert body["unit"] == "mas-evidence-backup.service" and body["test"] is False
|
|
assert "never delete retained objects" in body["action"]
|
|
|
|
|
|
def test_stale_notice_has_its_own_subject(tmp_path):
|
|
hub = Hub()
|
|
nt.notify(tmp_path / "n", unit="mas-evidence-freshness.service", sender=hub)
|
|
assert hub.sent[0]["subject"] == "MAS evidence backup stale"
|
|
|
|
|
|
def test_hub_outage_keeps_the_notice_and_replay_delivers_it_once(tmp_path):
|
|
state = tmp_path / "n"
|
|
down = nt.notify(state, unit="mas-evidence-backup.service", sender=Hub(down=True))
|
|
assert down["status"] == "pending_notification_requires_replay" and len(down["pending_event_ids"]) == 1
|
|
hub = Hub()
|
|
replay = nt.notify(state, sender=hub)
|
|
assert replay["status"] == "inbox_delivery_verified" and len(hub.sent) == 1
|
|
again = Hub()
|
|
nt.notify(state, sender=again)
|
|
assert again.sent == []
|
|
|
|
|
|
def test_test_unit_marks_the_notice_as_a_test(tmp_path):
|
|
hub = Hub()
|
|
nt.notify(tmp_path / "n", unit="mas-evidence-alert-test.service", sender=hub)
|
|
assert hub.sent[0]["subject"] == "MAS evidence notification test"
|
|
assert json.loads(hub.sent[0]["body"])["test"] is True
|
|
|
|
|
|
def test_unknown_unit_is_refused(tmp_path):
|
|
with pytest.raises(nt.Refused):
|
|
nt.notify(tmp_path / "n", unit="sshd.service", sender=Hub())
|
|
|
|
|
|
def test_echo_mismatch_is_not_counted_as_delivered(tmp_path):
|
|
def tampering(base, message):
|
|
raise nt.Refused("inbox_receipt_not_verified")
|
|
|
|
result = nt.notify(tmp_path / "n", unit="mas-evidence-backup.service", sender=tampering)
|
|
assert result["status"] == "pending_notification_requires_replay"
|
|
record = json.loads(next((tmp_path / "n").glob("*.json")).read_text())
|
|
assert record["status"] == "pending"
|