DataOps-platform/tests/integration/test_phase3_wp06_provision_outbox_postgres.py

"""Real PostgreSQL coverage for the provision outbox lease/fencing contract."""

from __future__ import annotations

import os
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 _runtime_url():
    value = os.environ.get("TEST_RUNTIME_DATABASE_URL")
    if not value:
        pytest.skip("TEST_RUNTIME_DATABASE_URL is required")
    return value


def _fixture(engine):
    actor, policy, grant, asset = (str(uuid.uuid4()) for _ in range(4))
    prefix = f"wp06-test-provision-{actor[:8]}"
    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})
        repository = SqlAlchemyTrustedDeliveryRepository(connection)
        repository.create_policy_version({
            "uid": policy, "code": f"WP06_PROVISION_{actor[:8]}", "version": "1",
            "status": "active", "selector": {}, "resource_rules": {},
            "created_by": actor, "current_version": 1,
        })
        repository.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": "2099-01-01T00:00:00+00:00", "status": "active",
            "reason_code": "approval_bound", "current_version": 1,
            "created_by": actor,
        })
    return actor, policy, grant, prefix


def _cleanup(engine, actor, policy, grant):
    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})


def test_provision_outbox_commits_before_lease_and_fences_retrying_workers():
    """A provider worker only runs while holding the DB-time current lease."""
    engine = create_engine(_url(), pool_pre_ping=True)
    actor = policy = grant = None
    digest = "b" * 64
    try:
        actor, policy, grant, prefix = _fixture(engine)

        def prepare(key):
            with engine.connect() as connection:
                repository = SqlAlchemyTrustedDeliveryRepository(connection)
                return repository.prepare_provision_operation(grant, key, digest)

        # Two different transport keys describe the same durable grant.  They
        # must converge on one outbox record before either worker can call out.
        with ThreadPoolExecutor(max_workers=2) as pool:
            first, second = list(pool.map(prepare, (f"{prefix}-one", f"{prefix}-two")))
        assert first["delivery_uid"] == second["delivery_uid"]

        with engine.connect() as observer:
            assert observer.execute(text("SELECT status FROM public.trusted_delivery_deliveries WHERE uid=CAST(:uid AS uuid)"), {"uid": first["delivery_uid"]}).scalar_one() == "pending"

        with engine.begin() as connection:
            repository = SqlAlchemyTrustedDeliveryRepository(connection)
            claimed_a = repository.claim_provision_operation(first["delivery_uid"], "worker-a")
            assert claimed_a["state"] == "claimed"
            assert repository.claim_provision_operation(first["delivery_uid"], "worker-b")["state"] == "leased"
            # A failed adapter is made pending while its live lease is still
            # fenced, instead of leaving a permanently processing record.
            repository.fail_provision_operation(first["delivery_uid"], "worker-a", claimed_a["lease_fence"])

        with engine.begin() as connection:
            repository = SqlAlchemyTrustedDeliveryRepository(connection)
            claimed_b = repository.claim_provision_operation(first["delivery_uid"], "worker-b")
            assert claimed_b["state"] == "claimed"
            assert claimed_b["lease_fence"] > claimed_a["lease_fence"]
            with pytest.raises(RuntimeError, match="lease conflict"):
                repository.complete_provision_operation(
                    grant, first["delivery_uid"], "worker-a", claimed_a["lease_fence"],
                    {"receipt_code": "LATE", "response_digest": "c" * 64, "diff_digest": "d" * 64},
                )
            result = repository.complete_provision_operation(
                grant, first["delivery_uid"], "worker-b", claimed_b["lease_fence"],
                {"receipt_code": "APPLIED", "response_digest": "c" * 64, "diff_digest": "d" * 64},
            )
            assert result["status"] == "applied"
            assert connection.execute(text("SELECT count(*) FROM public.trusted_delivery_receipts WHERE delivery_uid=CAST(:uid AS uuid)"), {"uid": first["delivery_uid"]}).scalar_one() == 1
            grant_state = connection.execute(text("SELECT status,reason_code,current_version FROM public.trusted_delivery_grants WHERE uid=CAST(:uid AS uuid)"), {"uid": grant}).one()
            assert grant_state == ("active", "provision_applied", 2)
    finally:
        if actor:
            _cleanup(engine, actor, policy, grant)
        engine.dispose()


def test_runtime_provision_gateway_fences_late_worker_and_is_visible_cross_connection(wp06_head):
    """The low-privilege runtime identity gets no direct DML escape hatch."""
    engine = create_engine(_url(), pool_pre_ping=True)
    runtime_engine = create_engine(_runtime_url(), pool_pre_ping=True)
    actor = policy = grant = None
    try:
        actor, policy, grant, prefix = _fixture(engine)
        with engine.begin() as connection:
            repository = SqlAlchemyTrustedDeliveryRepository(connection)
            operation = repository.prepare_provision_operation(grant, f"{prefix}-runtime", "e" * 64)

        with runtime_engine.begin() as connection:
            claimed_a = connection.execute(text("""SELECT public.trusted_delivery_runtime_write(
              'claim_provision', CAST(:payload AS jsonb))"""), {
                "payload": '{"uid":"' + operation["delivery_uid"] + '","worker":"runtime-a"}',
            }).scalar_one()
            assert claimed_a["state"] == "claimed"
            assert connection.execute(text("""SELECT public.trusted_delivery_runtime_write(
              'claim_provision', CAST(:payload AS jsonb))"""), {
                "payload": '{"uid":"' + operation["delivery_uid"] + '","worker":"runtime-b"}',
            }).scalar_one()["state"] == "leased"

        with engine.begin() as connection:
            connection.execute(text("""UPDATE public.trusted_delivery_deliveries
              SET lease_expires_at=clock_timestamp()-interval '1 second'
              WHERE uid=CAST(:uid AS uuid)"""), {"uid": operation["delivery_uid"]})

        with runtime_engine.begin() as connection:
            claimed_b = connection.execute(text("""SELECT public.trusted_delivery_runtime_write(
              'claim_provision', CAST(:payload AS jsonb))"""), {
                "payload": '{"uid":"' + operation["delivery_uid"] + '","worker":"runtime-b"}',
            }).scalar_one()
            assert claimed_b["state"] == "claimed"
            with pytest.raises(Exception, match="lease conflict"), connection.begin_nested():
                connection.execute(text("""SELECT public.trusted_delivery_runtime_write(
                  'complete_provision', CAST(:payload AS jsonb))"""), {
                    "payload": '{"grant_uid":"' + grant + '","uid":"' + operation["delivery_uid"] + '","worker":"runtime-a","fence":' + str(claimed_a["lease_fence"]) + ',"receipt_uid":"' + str(uuid.uuid4()) + '","receipt_code":"LATE","response_digest":"' + "f" * 64 + '","diff_digest":"' + "a" * 64 + '"}',
                })
            connection.execute(text("""SELECT public.trusted_delivery_runtime_write(
              'complete_provision', CAST(:payload AS jsonb))"""), {
                "payload": '{"grant_uid":"' + grant + '","uid":"' + operation["delivery_uid"] + '","worker":"runtime-b","fence":' + str(claimed_b["lease_fence"]) + ',"receipt_uid":"' + str(uuid.uuid4()) + '","receipt_code":"APPLIED","response_digest":"' + "f" * 64 + '","diff_digest":"' + "a" * 64 + '"}',
            }).scalar_one()

        with engine.connect() as observer:
            assert observer.execute(text("SELECT status FROM public.trusted_delivery_deliveries WHERE uid=CAST(:uid AS uuid)"), {"uid": operation["delivery_uid"]}).scalar_one() == "applied"
            assert observer.execute(text("SELECT reason_code FROM public.trusted_delivery_grants WHERE uid=CAST(:uid AS uuid)"), {"uid": grant}).scalar_one() == "provision_applied"
    finally:
        if actor:
            _cleanup(engine, actor, policy, grant)
        runtime_engine.dispose()
        engine.dispose()