from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] def test_retry_delay_is_exponential_and_capped(): from app.core.events.consumer import retry_delay assert retry_delay(1, base_seconds=2, maximum_seconds=30) == timedelta(seconds=2) assert retry_delay(2, base_seconds=2, maximum_seconds=30) == timedelta(seconds=4) assert retry_delay(10, base_seconds=2, maximum_seconds=30) == timedelta(seconds=30) def test_successful_dispatch_records_idempotent_completion(): from app.core.events.consumer import ClaimedEvent, dispatch_event event = ClaimedEvent( event_id="01900000-0000-7000-8000-000000000001", event_type="governance.updated", payload={"uid": "domain-1"}, attempts=0, ) calls = [] result = dispatch_event(event, lambda payload: calls.append(payload)) duplicate = dispatch_event( event, lambda payload: calls.append(payload), already_processed=True, ) assert result.status == "published" assert result.record_consumption is True assert duplicate.status == "published" assert duplicate.record_consumption is False assert calls == [{"uid": "domain-1"}] def test_failed_dispatch_retries_then_enters_dead_letter(): from app.core.events.consumer import ClaimedEvent, dispatch_event now = datetime(2026, 7, 16, tzinfo=timezone.utc) def fail(_payload): raise RuntimeError("downstream unavailable with secret-token") retry = dispatch_event( ClaimedEvent("event-1", "test", {}, attempts=0), fail, max_attempts=2, now=now, ) dead = dispatch_event( ClaimedEvent("event-1", "test", {}, attempts=1), fail, max_attempts=2, now=now, ) assert retry.status == "pending" assert retry.attempts == 1 assert retry.available_at > now assert retry.last_error == "downstream unavailable with [redacted]" assert dead.status == "dead_letter" assert dead.attempts == 2 def test_outbox_migration_and_claim_query_define_delivery_contract(): migration = ( ROOT / "migrations" / "versions" / "20260716_03_outbox.py" ).read_text(encoding="utf-8") repository = (ROOT / "app/core/events/outbox.py").read_text(encoding="utf-8") for column in ( "event_id", "aggregate_type", "aggregate_id", "event_type", "payload", "status", "attempts", "available_at", "published_at", "last_error", ): assert column in migration assert "outbox_consumptions" in migration assert "FOR UPDATE SKIP LOCKED" in repository assert "outbox_health" in repository def test_system_health_exposes_outbox_counts_without_payloads(): health = (ROOT / "app/core/system/health.py").read_text(encoding="utf-8") assert "outbox_health" in health assert 'health_status["outbox"]' in health assert "payload" not in health def test_consumer_without_registered_handlers_does_not_claim_events(monkeypatch): from app.commands import process_outbox def unexpected_claim(*_args, **_kwargs): raise AssertionError("events must not be claimed without a registered handler") monkeypatch.setattr(process_outbox, "claim_outbox", unexpected_claim) assert process_outbox.process_available_events(object(), handlers={}) == 0