tegwick
267d502d93
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m16s
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""Exercise interrupted/partial delivery without contacting live services."""
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SOURCE = Path(__file__).parents[1] / "scripts/backup_alerts.py"
|
|
spec = importlib.util.spec_from_file_location("backup_alerts_test", SOURCE)
|
|
alerts = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(alerts)
|
|
NOW = datetime(2026, 10, 31, 10, tzinfo=UTC)
|
|
EVENT = {"event_id": str(uuid.uuid4()), "condition": "backup_failed", "test": True}
|
|
|
|
|
|
def fail(*args):
|
|
raise TimeoutError("sensitive remote error must never be retained")
|
|
|
|
|
|
def email_ok(base, record, now):
|
|
return {"status": "resolution_accepted" if record.get("resolved_at")
|
|
else "active_email_route_verified", "accepted_at": now.isoformat(),
|
|
"email_delivery_proven": False}
|
|
|
|
|
|
def run(state, **kwargs):
|
|
options = {"now": NOW, "hub_sender": lambda *a: str(uuid.uuid4()),
|
|
"email_sender": email_ok}
|
|
options.update(kwargs)
|
|
return alerts.drain(state, "http://hub.invalid", "http://mail.invalid", **options)
|
|
|
|
|
|
def test_partial_failure_preserves_event_and_does_not_starve_other_channel(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
result = run(state, event=EVENT, hub_sender=fail)
|
|
assert result["failures"] == [{"event_id": EVENT["event_id"], "channel": "state_hub"}]
|
|
record = json.loads((state / (EVENT["event_id"] + ".json")).read_text())
|
|
assert record["alertmanager"]["status"] == "active_email_route_verified"
|
|
assert record["hub_message_id"] is None
|
|
assert "sensitive" not in json.dumps(record)
|
|
assert result["email_delivery_proven"] is False
|
|
assert run(state, event=EVENT)["status"] == "channel_handoffs_verified"
|
|
assert len(list(state.glob("*.json"))) == 1
|
|
|
|
|
|
def test_replay_does_not_repeat_confirmed_hub_but_refreshes_email_lease(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
run(state, event=EVENT, email_sender=fail)
|
|
calls = []
|
|
def email(base, record, now):
|
|
calls.append(record["event_id"])
|
|
return email_ok(base, record, now)
|
|
assert run(state, hub_sender=fail, email_sender=email)["failures"] == []
|
|
assert run(state, hub_sender=fail, email_sender=email)["failures"] == []
|
|
assert calls == [EVENT["event_id"], EVENT["event_id"]]
|
|
|
|
|
|
def test_resolution_retry_retains_original_event_and_stops_after_acceptance(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
run(state, event=EVENT)
|
|
later = NOW + timedelta(hours=1)
|
|
run(state, resolve=EVENT["event_id"], now=later, email_sender=fail)
|
|
path = state / (EVENT["event_id"] + ".json")
|
|
record = json.loads(path.read_text())
|
|
assert record["resolved_at"] == later.isoformat()
|
|
assert run(state, now=later, hub_sender=fail)["failures"] == []
|
|
assert run(state, now=later, hub_sender=fail, email_sender=fail)["failures"] == []
|
|
assert path.exists() # Resolution does not discard evidence.
|
|
|
|
|
|
def test_reused_id_and_unsafe_journal_refuse_before_send(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
run(state, event=EVENT)
|
|
with pytest.raises(alerts.Refused, match="reused"):
|
|
run(state, event={**EVENT, "condition": "backup_stale"})
|
|
path = state / (EVENT["event_id"] + ".json")
|
|
path.chmod(0o644)
|
|
with pytest.raises(alerts.Refused, match="unsafe"):
|
|
run(state, hub_sender=fail, email_sender=fail)
|
|
|
|
|
|
def test_full_journal_preserves_pending_and_existing_replay(tmp_path, monkeypatch):
|
|
state = tmp_path / "outbox"
|
|
monkeypatch.setattr(alerts, "MAX_EVENTS", 1)
|
|
run(state, event=EVENT, email_sender=fail)
|
|
with pytest.raises(alerts.Refused, match="journal_full"):
|
|
run(state, event={**EVENT, "event_id": str(uuid.uuid4())})
|
|
assert run(state)["failures"] == []
|
|
|
|
|
|
def test_hub_queue_receipt_is_not_delivery_and_exact_body_is_required(monkeypatch):
|
|
record = {**EVENT, "observed_at": NOW.isoformat()}
|
|
monkeypatch.setattr(alerts, "request_json", lambda *a: {"status": "queued", "id": "local"})
|
|
with pytest.raises(alerts.Refused, match="central_inbox"):
|
|
alerts.deliver_hub("http://hub.invalid", record)
|
|
expected = alerts.message(record)
|
|
uid = str(uuid.uuid4())
|
|
monkeypatch.setattr(alerts, "request_json", lambda *a: {**expected, "id": uid})
|
|
assert alerts.deliver_hub("http://hub.invalid", record) == uid
|
|
|
|
|
|
def test_email_route_readback_must_match_not_just_http_success(monkeypatch):
|
|
record = {**EVENT, "observed_at": NOW.isoformat()}
|
|
calls = []
|
|
receiver = "pending-operator-channel"
|
|
def request(url, payload=None):
|
|
calls.append((url, payload))
|
|
if payload is not None:
|
|
return None
|
|
return [{**calls[-2][1][0], "receivers": [{"name": receiver}]}]
|
|
monkeypatch.setattr(alerts, "request_json", request)
|
|
with pytest.raises(alerts.Refused, match="email_route"):
|
|
alerts.deliver_alertmanager("http://mail.invalid", record, NOW)
|
|
receiver = alerts.RECEIVER
|
|
result = alerts.deliver_alertmanager("http://mail.invalid", record, NOW)
|
|
assert result["email_delivery_proven"] is False
|
|
assert alerts.timestamp(calls[-2][1][0]["endsAt"]) == NOW + timedelta(hours=3)
|
|
assert calls[-2][1][0]["labels"]["event_id"] == EVENT["event_id"]
|
|
|
|
|
|
def test_native_api_millisecond_precision_is_used_for_exact_readback(monkeypatch):
|
|
now = NOW.replace(microsecond=123456)
|
|
record = {**EVENT, "observed_at": now.isoformat()}
|
|
posted = []
|
|
def request(url, payload=None):
|
|
if payload:
|
|
posted.extend(payload)
|
|
return None
|
|
return [{**posted[0], "receivers": [{"name": alerts.RECEIVER}]}]
|
|
monkeypatch.setattr(alerts, "request_json", request)
|
|
assert alerts.deliver_alertmanager("http://mail.invalid", record, now)["status"] == (
|
|
"active_email_route_verified")
|
|
assert posted[0]["startsAt"].endswith(".123+00:00")
|
|
assert posted[0]["endsAt"].endswith(".123+00:00")
|
|
|
|
|
|
def test_checkpoint_failure_before_first_send_leaves_no_unjournaled_send(tmp_path, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(alerts, "checkpoint", fail)
|
|
with pytest.raises(TimeoutError):
|
|
run(tmp_path / "outbox", event=EVENT, hub_sender=lambda *a: calls.append(a),
|
|
email_sender=lambda *a: calls.append(a))
|
|
assert calls == []
|
|
|
|
|
|
def test_health_episode_deduplicates_then_resolves_and_recurrence_is_new(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
observations = {c: c == "pending_backup" for c in alerts.CONDITIONS}
|
|
run(state, observations=observations)
|
|
run(state, observations=observations, now=NOW + timedelta(hours=1))
|
|
assert len(list(state.glob("*.json"))) == 1
|
|
run(state, observations=dict.fromkeys(alerts.CONDITIONS, False),
|
|
now=NOW + timedelta(hours=2))
|
|
first = json.loads(next(state.glob("*.json")).read_text())
|
|
assert first["alertmanager"]["status"] == "resolution_accepted"
|
|
run(state, observations=observations, now=NOW + timedelta(hours=3))
|
|
assert len(list(state.glob("*.json"))) == 2
|
|
|
|
|
|
def test_service_failure_needs_a_newer_healthy_paired_capture_to_clear(tmp_path):
|
|
state = tmp_path / "outbox"
|
|
event = {**EVENT, "test": False}
|
|
run(state, event=event)
|
|
healthy = dict.fromkeys(alerts.CONDITIONS, False)
|
|
path = state / (EVENT["event_id"] + ".json")
|
|
run(state, observations=healthy, recovered_capture=(NOW - timedelta(hours=1)).isoformat())
|
|
assert "resolved_at" not in json.loads(path.read_text())
|
|
run(state, observations=healthy, recovered_capture=(NOW + timedelta(minutes=1)).isoformat(),
|
|
now=NOW + timedelta(hours=1))
|
|
assert json.loads(path.read_text())["alertmanager"]["status"] == "resolution_accepted"
|
|
|
|
|
|
def test_unknown_health_and_future_capture_cannot_clear_incidents(tmp_path):
|
|
with pytest.raises(alerts.Refused, match="complete_health"):
|
|
run(tmp_path / "outbox", observations={"backup_stale": False})
|
|
with pytest.raises(alerts.Refused, match="healthy_new_capture"):
|
|
run(tmp_path / "outbox", observations=dict.fromkeys(alerts.CONDITIONS, False),
|
|
recovered_capture=(NOW + timedelta(hours=1)).isoformat())
|
|
|
|
|
|
def cli_module(monkeypatch, name):
|
|
monkeypatch.syspath_prepend(str(SOURCE.parent))
|
|
module_spec = importlib.util.spec_from_file_location(name, SOURCE.with_name(name + ".py"))
|
|
module = importlib.util.module_from_spec(module_spec)
|
|
module_spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_journal_pressure_emits_failure_before_full_without_claiming_recovery(
|
|
tmp_path, monkeypatch, capsys,
|
|
):
|
|
monitor = cli_module(monkeypatch, "observe-backup-alerts")
|
|
monkeypatch.setattr(monitor, "MAX_EVENTS", 4)
|
|
for i in range(3):
|
|
(tmp_path / f"{i}.json").write_text("synthetic")
|
|
monkeypatch.setattr(monitor.health, "check", lambda *a, **kw: {
|
|
"status": "healthy", "reasons": [], "captured_at": NOW.isoformat()})
|
|
calls = []
|
|
def drain(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
return {"status": "channel_handoffs_verified"}
|
|
monkeypatch.setattr(monitor, "drain", drain)
|
|
monkeypatch.setattr(sys, "argv", ["observe", "--execute", "--state-dir", str(tmp_path),
|
|
"--mirror-receipt", str(tmp_path / "unused"), "--alert-state-dir", str(tmp_path),
|
|
"--minimum-free-bytes", "1", "--maximum-state-bytes", "1",
|
|
"--hub-base", "http://hub.invalid", "--alertmanager-base", "http://mail.invalid"])
|
|
assert monitor.main() == 0 # Monitoring succeeded, while backup health is unhealthy.
|
|
assert calls[0]["observations"]["backup_failed"] is True
|
|
assert calls[0]["recovered_capture"] is None
|
|
assert json.loads(capsys.readouterr().out)["health"]["status"] == "unhealthy"
|
|
|
|
|
|
def test_failure_handler_requires_matching_systemd_invocation(tmp_path, monkeypatch):
|
|
handler = cli_module(monkeypatch, "notify-backup-alert")
|
|
unit = "coulomb-social-paired-backup.service"
|
|
monkeypatch.setattr(sys, "argv", ["notify", "--execute", "--failed-unit", unit,
|
|
"--state-dir", str(tmp_path), "--hub-base", "http://hub.invalid",
|
|
"--alertmanager-base", "http://mail.invalid"])
|
|
monkeypatch.setenv("MONITOR_UNIT", "other.service")
|
|
monkeypatch.setenv("MONITOR_INVOCATION_ID", EVENT["event_id"].replace("-", ""))
|
|
monkeypatch.setenv("MONITOR_SERVICE_RESULT", "exit-code")
|
|
with pytest.raises(SystemExit):
|
|
handler.main()
|
|
calls = []
|
|
def drain(*a, **kw):
|
|
calls.append(kw)
|
|
return {"status": "channel_handoffs_verified"}
|
|
monkeypatch.setattr(handler, "drain", drain)
|
|
monkeypatch.setenv("MONITOR_UNIT", unit)
|
|
assert handler.main() == 0
|
|
assert calls[0]["event"] == {**EVENT, "event_id": EVENT["event_id"].replace("-", ""),
|
|
"test": False}
|