DataOps-platform/tests/test_outbox.py

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