DataOps-platform/tests/integration/test_phase3_wp06_subscription_postgres.py

from __future__ import annotations

import os
import subprocess
import uuid
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text

pytestmark = pytest.mark.integration
ROOT = Path(__file__).resolve().parents[2]


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


def _upgrade_head():
    environment = dict(os.environ)
    environment["MIGRATION_DATABASE_URL"] = os.environ.get("TEST_MIGRATION_DATABASE_URL", _url())
    subprocess.run(
        [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), "upgrade", "head"],
        cwd=ROOT, env=environment, check=True, capture_output=True, text=True,
    )


def test_subscription_postgres_uses_db_time_fencing_and_bounded_dead_letter():
    from app.core.system.trusted_delivery_subscription_repository import (
        SqlAlchemyTrustedDeliverySubscriptionRepository,
    )

    _upgrade_head()
    engine = create_engine(_url(), pool_pre_ping=True)
    actor, policy, grant, subscription, asset = (str(uuid.uuid4()) for _ in range(5))
    key = f"wp06-test-sub-{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": key})
            connection.execute(text("INSERT INTO public.trusted_delivery_policy_versions(uid,code,version,status,selector,resource_rules,created_by) VALUES(CAST(:uid AS uuid),:code,'1','active','{}','{}',CAST(:actor AS uuid))"), {"uid": policy, "code": f"WP06_SUB_{actor[:8]}", "actor": actor})
            connection.execute(text("""INSERT INTO public.trusted_delivery_policy_transitions(uid,policy_uid,operation,idempotency_key,request_digest,approval_ref,approval_digest,actor_uid)
              VALUES(CAST(:uid AS uuid),CAST(:policy AS uuid),'activate',:key,:digest,'approval:fixture',:digest,CAST(:actor AS uuid))"""), {"uid": str(uuid.uuid4()), "policy": policy, "key": f"{key}-policy", "digest": "a" * 64, "actor": actor})
            connection.execute(text("""INSERT INTO public.trusted_delivery_grants(uid,policy_uid,subject_uid,asset_uid,provider,idempotency_key,request_digest,expires_at,status,reason_code,current_version,created_by,approval_ref,approval_digest,purpose,environment,target_capability)
              VALUES(CAST(:uid AS uuid),CAST(:policy AS uuid),CAST(:actor AS uuid),CAST(:asset AS uuid),'database',:key,:digest,'2026-12-31T00:00:00+00:00','active','test',1,CAST(:actor AS uuid),'approval:fixture',:digest,'quality_review','test','{}')"""), {"uid": grant, "policy": policy, "actor": actor, "asset": asset, "key": f"{key}-grant", "digest": "a" * 64})
            repository = SqlAlchemyTrustedDeliverySubscriptionRepository(connection)
            created = repository.create_subscription({"uid": subscription, "grant_uid": grant, "asset_uid": asset, "trigger": {"kind": "schedule", "schedule_ref": "wp06-test-schedule"}, "purpose": "quality_review", "expires_at": "2026-12-30T00:00:00+00:00", "idempotency_key": key, "request_digest": "b" * 64, "status": "draft", "current_version": 1, "created_by": actor})
            active = repository.transition_subscription(created["uid"], {"draft"}, "active")
            delivery = repository.enqueue_subscription_delivery({"uid": str(uuid.uuid4()), "subscription_uid": active["uid"], "trigger_ref": "wp06-test-trigger", "event_digest": "c" * 64, "idempotency_key": f"{key}-delivery", "request_digest": "d" * 64, "status": "pending", "attempt_count": 0, "lease_owner": None, "lease_fence": 0})
        with engine.begin() as first:
            claimed = SqlAlchemyTrustedDeliverySubscriptionRepository(first).claim_subscription_delivery(delivery["uid"], "worker-one")
            assert claimed["lease_fence"] == 1
        with engine.begin() as second:
            assert SqlAlchemyTrustedDeliverySubscriptionRepository(second).claim_subscription_delivery(delivery["uid"], "worker-two") is None
        for attempt in range(3):
            with engine.begin() as connection:
                repository = SqlAlchemyTrustedDeliverySubscriptionRepository(connection)
                current = claimed if attempt == 0 else repository.claim_subscription_delivery(delivery["uid"], "worker-one")
                if current is None:
                    connection.execute(text("UPDATE public.trusted_delivery_subscription_deliveries SET next_attempt_at=clock_timestamp()-interval '1 second' WHERE uid=CAST(:uid AS uuid)"), {"uid": delivery["uid"]})
                    current = repository.claim_subscription_delivery(delivery["uid"], "worker-one")
                result = repository.record_subscription_attempt(delivery["uid"], "worker-one", current["lease_fence"], False)
        assert result["status"] == "dead_letter"
        with engine.begin() as connection:
            repository = SqlAlchemyTrustedDeliverySubscriptionRepository(connection)
            compensated = repository.compensate_subscription_delivery(delivery["uid"], "operator_review", "COMPENSATED")
            assert compensated["status"] == "compensated"
    finally:
        with engine.begin() as connection:
            connection.execute(text("DELETE FROM public.trusted_delivery_subscription_deliveries WHERE idempotency_key LIKE 'wp06-test-sub-%'"))
            connection.execute(text("DELETE FROM public.trusted_delivery_subscriptions WHERE idempotency_key LIKE 'wp06-test-sub-%'"))
            connection.execute(text("DELETE FROM public.trusted_delivery_grants WHERE idempotency_key LIKE 'wp06-test-sub-%'"))
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_transitions WHERE idempotency_key LIKE 'wp06-test-sub-%'"))
            connection.execute(text("DELETE FROM public.trusted_delivery_policy_versions WHERE code LIKE 'WP06_SUB_%'"))
            connection.execute(text("DELETE FROM public.users WHERE username LIKE 'wp06-test-sub-%'"))
        engine.dispose()