tegwick
93b6ba2aae
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
67 lines
2.8 KiB
Python
67 lines
2.8 KiB
Python
"""Inbox notification failure/replay with synthetic receipts; no live messages."""
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def notifier(monkeypatch):
|
|
scripts = Path(__file__).resolve().parents[1] / "scripts"
|
|
monkeypatch.syspath_prepend(str(scripts))
|
|
spec = importlib.util.spec_from_file_location("ledger_notifier", scripts / "notify-ledger-failure.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_outage_preserves_private_pending_notice_and_replay_delivers(notifier, tmp_path):
|
|
state = tmp_path / "outbox"
|
|
def down(base, message):
|
|
raise OSError("synthetic HTTP failure")
|
|
with pytest.raises(OSError):
|
|
notifier.notify(state, "http://synthetic.invalid", next(iter(notifier.UNITS)), sender=down)
|
|
path, = state.glob("*.json")
|
|
pending = json.loads(path.read_text())
|
|
assert pending["status"] == "pending"
|
|
assert state.stat().st_mode & 0o777 == 0o700
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
sent = []
|
|
def accepted(base, message):
|
|
sent.append(message)
|
|
return "00000000-0000-0000-0000-000000000001"
|
|
receipt = notifier.notify(state, "http://synthetic.invalid", sender=accepted)
|
|
assert receipt["recipient"] == "platform-root"
|
|
assert receipt["human_acknowledged"] is False
|
|
assert json.loads(path.read_text())["status"] == "delivered"
|
|
assert json.loads(sent[0]["body"])["event_id"] == pending["event_id"]
|
|
notifier.notify(state, "http://synthetic.invalid", sender=accepted)
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_unexpected_unit_is_rejected_before_outbox_mutation(notifier, tmp_path):
|
|
with pytest.raises(notifier.Refused, match="invalid_notification_unit"):
|
|
notifier.notify(tmp_path / "unused", "http://synthetic.invalid", "other.service")
|
|
assert not (tmp_path / "unused").exists()
|
|
|
|
|
|
def test_failure_hook_test_is_labelled_without_real_failure_claim(notifier, tmp_path):
|
|
sent = []
|
|
def accepted(base, message):
|
|
sent.append(message)
|
|
return "00000000-0000-0000-0000-000000000001"
|
|
notifier.notify(tmp_path / "outbox", "http://synthetic.invalid", notifier.TEST_UNIT, sender=accepted)
|
|
assert json.loads(sent[0]["body"])["test"] is True
|
|
assert sent[0]["subject"].endswith("notification test")
|
|
|
|
|
|
def test_queued_response_is_not_central_inbox_delivery(notifier, monkeypatch):
|
|
class Response:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def read(self, *args): return b'{"status":"queued","receipt_id":"synthetic"}'
|
|
monkeypatch.setattr(notifier, "urlopen", lambda *args, **kwargs: Response())
|
|
with pytest.raises(notifier.Refused, match="inbox_receipt_not_verified"):
|
|
notifier.post("http://synthetic.invalid", {"subject": "synthetic", "body": "synthetic"})
|