"""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()