DataOps-platform/tests/integration/test_phase3_wp06_reclaim_outbox_postgres.py

from __future__ import annotations

import os
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor

import pytest
from sqlalchemy import create_engine, text

from app.core.system.trusted_delivery_repository import (
    SqlAlchemyTrustedDeliveryRepository,
)

pytestmark = pytest.mark.integration


def _url():
    value = os.environ.get("TEST_DATABASE_URL")
    if not value:
        pytest.skip("TEST_DATABASE_URL is required")
    return value


def test_reclaim_outbox_is_fenced_replay_safe_and_restart_recoverable():
    """Two DB connections exercise the durable revoke outbox, not a mock lock."""
    engine = create_engine(_url(), pool_pre_ping=True)
    actor, policy, grant, asset = (str(uuid.uuid4()) for _ in range(4))
    prefix = f"wp06-test-outbox-{actor[:8]}"
    try:
        with engine.begin() as connection:
            connection.execute(text("INSERT INTO public.users(id,username,display_name,password_hash,status) VALUES(CAST(:id AS uuid),:name,:name,'test','active')"), {"id": actor, "name": prefix})
            repo = SqlAlchemyTrustedDeliveryRepository(connection)
            repo.create_policy_version({"uid": policy, "code": f"WP06_OUTBOX_{actor[:8]}", "version": "1", "status": "active", "selector": {}, "resource_rules": {}, "created_by": actor, "current_version": 1})
            repo.enqueue_grant({"uid": grant, "policy_uid": policy, "subject_uid": actor, "asset_uid": asset, "provider": "database", "idempotency_key": f"{prefix}-grant", "request_digest": "a" * 64, "expires_at": "2026-08-01T00:00:00+00:00", "status": "active", "reason_code": "fixture", "current_version": 1, "created_by": actor})

        barrier = threading.Barrier(2)
        calls = []

        def claim(key):
            with engine.begin() as connection:
                barrier.wait(timeout=5)
                result = SqlAlchemyTrustedDeliveryRepository(connection).claim_reclaim_operation(grant, key, "b" * 64, f"worker-{key}")
                if result["state"] == "claimed":
                    calls.append(result["delivery_uid"])
                return result

        with ThreadPoolExecutor(max_workers=2) as pool:
            first, second = list(pool.map(claim, ("one", "two")))
        claimed = next(item for item in (first, second) if item["state"] == "claimed")
        assert sorted(item["state"] for item in (first, second)) == ["claimed", "leased"]
        assert calls == [claimed["delivery_uid"]]  # provider may be called only by claimant

        with engine.begin() as connection:
            repo = SqlAlchemyTrustedDeliveryRepository(connection)
            assert repo.claim_reclaim_operation(grant, "one", "b" * 64, "worker-replay")["state"] == "leased"
            with pytest.raises(RuntimeError, match="idempotency conflict"):
                repo.claim_reclaim_operation(grant, "one", "c" * 64, "worker-conflict")
            connection.execute(text("UPDATE public.trusted_delivery_deliveries SET lease_expires_at=clock_timestamp()-interval '1 second' WHERE uid=CAST(:uid AS uuid)"), {"uid": claimed["delivery_uid"]})

        with engine.begin() as connection:
            repo = SqlAlchemyTrustedDeliveryRepository(connection)
            restarted = repo.claim_reclaim_operation(grant, "two", "b" * 64, "worker-restarted")
            assert restarted["state"] == "claimed"
            with pytest.raises(RuntimeError, match="lease conflict"):
                repo.complete_claimed_reclaim(grant, claimed["delivery_uid"], "worker-one", claimed["lease_fence"], {"receipt_code": "LATE", "response_digest": "d" * 64, "diff_digest": "e" * 64})
            repo.fail_reclaim_operation(restarted["delivery_uid"], "worker-restarted", restarted["lease_fence"])

        with engine.begin() as connection:
            repo = SqlAlchemyTrustedDeliveryRepository(connection)
            retry = repo.claim_reclaim_operation(grant, "two", "b" * 64, "worker-retry")
            assert retry["state"] == "claimed"
            saved = repo.complete_claimed_reclaim(grant, retry["delivery_uid"], "worker-retry", retry["lease_fence"], {"receipt_code": "REVOKED", "response_digest": "d" * 64, "diff_digest": "e" * 64})
            assert saved["status"] == "reclaimed"
            assert connection.execute(text("SELECT count(*) FROM public.trusted_delivery_receipts WHERE grant_uid=CAST(:uid AS uuid)"), {"uid": grant}).scalar_one() == 1
    finally:
        with engine.begin() as connection:
            connection.execute(text("ALTER TABLE public.trusted_delivery_receipts DISABLE TRIGGER USER"))
            connection.execute(text("DELETE FROM public.trusted_delivery_receipts WHERE grant_uid=CAST(:uid AS uuid)"), {"uid": grant})
            connection.execute(text("ALTER TABLE public.trusted_delivery_receipts ENABLE TRIGGER USER"))
            connection.execute(text("DELETE FROM public.trusted_delivery_deliveries WHERE grant_uid=CAST(:uid AS uuid)"), {"uid": grant})
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_transitions WHERE policy_uid=CAST(:uid AS uuid)"), {"uid": policy})
            connection.execute(text("DELETE FROM public.trusted_delivery_grants WHERE uid=CAST(:uid AS uuid)"), {"uid": grant})
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_versions WHERE uid=CAST(:uid AS uuid)"), {"uid": policy})
            connection.execute(text("DELETE FROM public.users WHERE id=CAST(:uid AS uuid)"), {"uid": actor})
        engine.dispose()