DataOps-platform/tests/integration/test_wp10_tenant_postgres.py

from __future__ import annotations

import os
import subprocess
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.engine import make_url
from sqlalchemy.exc import SQLAlchemyError

from app.core.system.tenant_repository import SqlAlchemyTenantFoundationRepository

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


def test_wp10_real_postgres_runtime_is_rls_scoped_and_two_connections_cannot_overreserve():
    # Superseded by claim-bound tests below: direct runtime repository access
    # must now fail before any payload or tenant GUC can reach PostgreSQL.
    with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
        SqlAlchemyTenantFoundationRepository(object())
    return
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
    engine = create_engine(database_url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            repository = SqlAlchemyTenantFoundationRepository(connection)
            repository.provision(tenant_id, "private_single_tenant", "provision-1")
        first = engine.connect()
        second = engine.connect()
        first_transaction = first.begin()
        second_transaction = second.begin()
        try:
            first_repo = SqlAlchemyTenantFoundationRepository(first)
            second_repo = SqlAlchemyTenantFoundationRepository(second)
            fence = first_repo.reserve_quota(tenant_id, "records", 1, "reserve-1")
            first_transaction.commit()
            with pytest.raises(ValueError, match="tenant_quota_exhausted"):
                second_repo.reserve_quota(tenant_id, "records", 1, "reserve-2")
            second_transaction.rollback()
            with engine.begin() as check:
                runtime = SqlAlchemyTenantFoundationRepository(check)
                assert runtime.list_audit(tenant_id)[0]["tenant_id"] == tenant_id
                with pytest.raises(ValueError, match="tenant_scope_denied"):
                    runtime.list_audit("wrong-tenant")
                assert fence == 1
        finally:
            if first_transaction.is_active:
                first_transaction.rollback()
            if second_transaction.is_active:
                second_transaction.rollback()
            first.close()
            second.close()
    finally:
        with engine.begin() as cleanup:
            cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        engine.dispose()


def test_wp10_runtime_role_cannot_directly_mutate_tenant_audit_after_role_init():
    with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
        SqlAlchemyTenantFoundationRepository(object())
    return
    database_url = os.getenv("TEST_DATABASE_URL")
    runtime_url = os.getenv("TEST_RUNTIME_DATABASE_URL")
    if not database_url or not runtime_url:
        pytest.skip("TEST_DATABASE_URL and TEST_RUNTIME_DATABASE_URL are required")
    tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
    admin = create_engine(database_url, pool_pre_ping=True)
    runtime = create_engine(runtime_url, pool_pre_ping=True)
    try:
        with admin.begin() as connection:
            SqlAlchemyTenantFoundationRepository(connection).provision(
                tenant_id, "private_single_tenant", "provision-1"
            )
        with runtime.begin() as connection:
            connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            with pytest.raises(SQLAlchemyError):
                connection.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
    finally:
        with admin.begin() as cleanup:
            cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        runtime.dispose()
        admin.dispose()


def test_wp10_lifecycle_fence_replay_and_frozen_task_rejection_persist_in_postgres():
    with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
        SqlAlchemyTenantFoundationRepository(object())
    return
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
    engine = create_engine(database_url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            repository = SqlAlchemyTenantFoundationRepository(connection)
            repository.provision(tenant_id, "private_single_tenant", "provision-1")
            frozen = repository.transition_lifecycle(tenant_id, "freeze", 0, "freeze-1")
            replay = repository.transition_lifecycle(tenant_id, "freeze", 0, "freeze-1")
            assert replay == frozen
            with pytest.raises(ValueError, match="tenant_fence_conflict"):
                repository.transition_lifecycle(tenant_id, "begin_recovery", 0, "recover-1", approval_ref="approval-1")
            with pytest.raises(ValueError, match="tenant_not_active"):
                repository.reserve_quota(tenant_id, "background_tasks", 1, "task-while-frozen")
            recovered = repository.transition_lifecycle(tenant_id, "begin_recovery", frozen, "recover-1", approval_ref="approval-1")
            active = repository.transition_lifecycle(tenant_id, "activate", recovered, "activate-1", approval_ref="approval-1")
            assert active > recovered > frozen
    finally:
        with engine.begin() as cleanup:
            cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_lifecycle_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        engine.dispose()


def test_wp10_postgres_quota_accepts_bounded_decimal_without_overreserve():
    with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
        SqlAlchemyTenantFoundationRepository(object())
    return
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
    engine = create_engine(database_url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            repository = SqlAlchemyTenantFoundationRepository(connection)
            repository.provision(tenant_id, "private_single_tenant", "provision-1")
            assert repository.reserve_quota(tenant_id, "records", 0.5, "decimal-1") == 1
            with pytest.raises(ValueError, match="tenant_quota_exhausted"):
                repository.reserve_quota(tenant_id, "records", 0.75, "decimal-2")
    finally:
        with engine.begin() as cleanup:
            cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        engine.dispose()


def test_wp10_runtime_cannot_claim_another_tenant_by_payload_or_guc():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    alpha = f"wp10-alpha-{uuid.uuid4().hex[:12]}"
    beta = f"wp10-beta-{uuid.uuid4().hex[:12]}"
    runtime_name = f"wp10_runtime_{uuid.uuid4().hex[:12]}"
    runtime_password = f"wp10_{uuid.uuid4().hex}"
    admin = create_engine(database_url, pool_pre_ping=True)
    runtime = None
    try:
        with admin.begin() as connection:
            quoted_runtime = connection.dialect.identifier_preparer.quote(runtime_name)
            connection.execute(text(f"CREATE ROLE {quoted_runtime} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": runtime_password})
            connection.execute(text(f"GRANT dataops_app_runtime TO {quoted_runtime}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": alpha})
        runtime = create_engine(make_url(database_url).set(username=runtime_name, password=runtime_password).render_as_string(hide_password=False), pool_pre_ping=True)
        with runtime.begin() as connection:
            connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": alpha})
            with pytest.raises(SQLAlchemyError, match="permission denied"):
                connection.execute(text("SELECT public.tenant_foundation_write('provision', CAST(:payload AS jsonb))"), {"payload": f'{{"tenant_id":"{beta}","delivery_mode":"private_single_tenant","idempotency_key":"beta-provision"}}'})
        with admin.connect() as connection:
            assert connection.execute(text("SELECT count(*) FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": beta}).scalar_one() == 0
    finally:
        with admin.begin() as cleanup:
            for tenant_id in (alpha, beta):
                cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
                cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
                cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
                cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        if runtime is not None:
            runtime.dispose()
        with admin.begin() as connection:
            connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(runtime_name)}"))
        admin.dispose()


def test_wp10_control_claim_is_membership_derived_one_shot_and_runtime_cannot_issue():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-claim-{uuid.uuid4().hex[:12]}"
    principal = str(uuid.uuid4())
    claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
    control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
    control_password = f"wp10_{uuid.uuid4().hex}"
    runtime_name = f"wp10_runtime_{uuid.uuid4().hex[:12]}"
    runtime_password = f"wp10_{uuid.uuid4().hex}"
    admin = create_engine(database_url, pool_pre_ping=True)
    runtime = None
    control = None
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(
                f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
            ), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
            quoted_runtime = connection.dialect.identifier_preparer.quote(runtime_name)
            connection.execute(text(
                f"CREATE ROLE {quoted_runtime} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
            ), {"password": runtime_password})
            connection.execute(text(f"GRANT dataops_app_runtime TO {quoted_runtime}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
        control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
        control = create_engine(control_url, pool_pre_ping=True)
        runtime = create_engine(make_url(database_url).set(username=runtime_name, password=runtime_password).render_as_string(hide_password=False), pool_pre_ping=True)
        with control.begin() as connection:
            assert connection.execute(text("SELECT current_user")).scalar_one() == control_name
            issued = connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"quota_reserve","request":{{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}}}}'}).scalar_one()
            assert dict(issued)["tenant_id"] == tenant_id
        with runtime.begin() as connection, pytest.raises(SQLAlchemyError):
            connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": "{}"})
        with control.begin() as connection:
            assert dict(connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": '{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}'}).scalar_one())["tenant_id"] == tenant_id
            with pytest.raises(SQLAlchemyError):
                connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": '{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}'})
    finally:
        with admin.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            # The append-only tenant audit table is RLS-scoped even for the
            # owner role.  Test cleanup must declare the exact generated
            # tenant scope; it must not weaken RLS or use a broad delete.
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        if control is not None:
            control.dispose()
        if runtime is not None:
            runtime.dispose()
        with admin.begin() as connection:
            connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(control_name)}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(runtime_name)}"))
        admin.dispose()


def test_wp10_control_claim_two_connections_allow_one_consumer_and_reject_after_restart():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-claim-race-{uuid.uuid4().hex[:10]}"
    principal = str(uuid.uuid4())
    claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
    request_json = '{"quota_name":"records","amount":"1","idempotency_key":"claim-race-1"}'
    control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
    control_password = f"wp10_{uuid.uuid4().hex}"
    admin = create_engine(database_url, pool_pre_ping=True)
    control = None
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(
                f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
            ), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
        control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
        control = create_engine(control_url, pool_pre_ping=True)
        with control.begin() as connection:
            connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"quota_reserve","request":{request_json}}}'})

        barrier = threading.Barrier(2)

        def consume_once() -> bool:
            with control.begin() as connection:
                barrier.wait(timeout=10)
                try:
                    connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": request_json}).scalar_one()
                    return True
                except SQLAlchemyError:
                    return False

        with ThreadPoolExecutor(max_workers=2) as workers:
            outcomes = list(workers.map(lambda _: consume_once(), range(2)))
        assert outcomes.count(True) == 1
        assert outcomes.count(False) == 1

        control.dispose()
        control = create_engine(control_url, pool_pre_ping=True)
        with control.begin() as restarted, pytest.raises(SQLAlchemyError, match="tenant_claim_replayed_or_expired"):
            restarted.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": request_json})
    finally:
        if control is not None:
            control.dispose()
        with admin.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
        admin.dispose()


def test_wp10_control_claim_binds_decimal_quota_reserve_and_release_atomically():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-quota-{uuid.uuid4().hex[:14]}"
    principal = str(uuid.uuid4())
    control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
    control_password = f"wp10_{uuid.uuid4().hex}"
    request_json = '{"quota_name":"records","amount":"0.5","idempotency_key":"quota-1"}'
    admin = create_engine(database_url, pool_pre_ping=True)
    control = None
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_quotas(tenant_id,quota_name,limit_units) VALUES(:tenant_id,'records',1)"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
        control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
        control = create_engine(control_url, pool_pre_ping=True)

        def claim(action: str, request: str) -> tuple[str, str]:
            claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
            with control.begin() as connection:
                connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"{action}","request":{request}}}'})
            return claim_uid, nonce

        reserve_claim, reserve_nonce = claim("quota_reserve", request_json)
        with control.begin() as connection:
            reserved = connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": reserve_claim, "nonce": reserve_nonce, "action": "quota_reserve", "request": request_json}).scalar_one()
            assert dict(reserved)["status"] == "reserved"
        with control.begin() as connection, pytest.raises(SQLAlchemyError):
            connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": reserve_claim, "nonce": reserve_nonce, "action": "quota_reserve", "request": request_json})
        release_request = '{"quota_name":"records","idempotency_key":"quota-1","lease_fence":"1"}'
        release_claim, release_nonce = claim("quota_release", release_request)
        with control.begin() as connection:
            released = connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": release_claim, "nonce": release_nonce, "action": "quota_release", "request": release_request}).scalar_one()
            assert dict(released)["status"] == "released"
        with admin.connect() as check:
            assert check.execute(text("SELECT reserved_units FROM public.tenant_quotas WHERE tenant_id=:tenant_id AND quota_name='records'"), {"tenant_id": tenant_id}).scalar_one() == 0
        settle_request = '{"quota_name":"records","amount":"0.25","idempotency_key":"settle-1"}'
        settle_reserve_claim, settle_reserve_nonce = claim("quota_reserve", settle_request)
        with control.begin() as connection:
            assert dict(connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": settle_reserve_claim, "nonce": settle_reserve_nonce, "action": "quota_reserve", "request": settle_request}).scalar_one())["status"] == "reserved"
        settle_request = '{"quota_name":"records","idempotency_key":"settle-1","lease_fence":"3"}'
        settle_claim, settle_nonce = claim("quota_settle", settle_request)
        with control.begin() as connection:
            assert dict(connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": settle_claim, "nonce": settle_nonce, "action": "quota_settle", "request": settle_request}).scalar_one())["status"] == "settled"
        late_claim, late_nonce = claim("quota_release", settle_request)
        with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_quota_terminal"):
            connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": late_claim, "nonce": late_nonce, "action": "quota_release", "request": settle_request})
        invalid_request = '{"quota_name":"records","amount":1.0,"idempotency_key":"float-1"}'
        invalid_claim, invalid_nonce = claim("quota_reserve", invalid_request)
        with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_payload_closed"):
            connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": invalid_claim, "nonce": invalid_nonce, "action": "quota_reserve", "request": invalid_request})
        overflow_request = '{"quota_name":"records","amount":"1000000000000000000000001","idempotency_key":"overflow-1"}'
        overflow_claim, overflow_nonce = claim("quota_reserve", overflow_request)
        with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_payload_invalid"):
            connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": overflow_claim, "nonce": overflow_nonce, "action": "quota_reserve", "request": overflow_request})
    finally:
        if control is not None:
            control.dispose()
        with admin.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
        admin.dispose()


def test_wp10_unapproved_shared_delivery_mode_cannot_issue_control_claim():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id, principal = f"wp10-shared-{uuid.uuid4().hex[:12]}", str(uuid.uuid4())
    control_name, control_password = f"wp10_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
    claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
    admin = create_engine(database_url, pool_pre_ping=True)
    control = None
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'shared_control_plane','active')"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'shared.example')"), {"tenant_id": tenant_id, "principal": principal})
        control = create_engine(make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False), pool_pre_ping=True)
        with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_membership_denied"):
            connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"shared.example","action":"quota_reserve","request":{{"quota_name":"records","amount":"1","idempotency_key":"shared-1"}}}}'})
    finally:
        if control is not None:
            control.dispose()
        with admin.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
        admin.dispose()


def test_wp10_claim_bound_lifecycle_fence_hold_backup_rollback_and_delete():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    control_name, control_password = f"wp10_legacy_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
    admin = create_engine(database_url, pool_pre_ping=True)
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
        control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
        with create_engine(control_url).connect() as connection, pytest.raises(SQLAlchemyError):
            connection.execute(text("SELECT public.tenant_control_lifecycle_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),'{}'::jsonb)"), {"claim_uid": str(uuid.uuid4()), "nonce": str(uuid.uuid4())})
    finally:
        with admin.begin() as connection:
            connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(control_name)}"))
        admin.dispose()
    return
    tenant_id, principal = f"wp10-life-{uuid.uuid4().hex[:12]}", str(uuid.uuid4())
    control_name, control_password = f"wp10_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
    admin = create_engine(database_url, pool_pre_ping=True)
    control = None
    try:
        with admin.begin() as connection:
            quoted_control = connection.dialect.identifier_preparer.quote(control_name)
            connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
            connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
        control = create_engine(make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False), pool_pre_ping=True)

        def transition(operation: str, fence: int, key: str, *, approval: str | None = None, backup: str | None = None, retention: str | None = None):
            request = {"operation": operation, "expected_fence": str(fence), "idempotency_key": key, "approval_ref": approval, "backup_digest": backup, "retention_seconds": retention}
            claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
            with control.begin() as connection:
                connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": __import__('json').dumps({"claim_uid": claim_uid, "nonce": nonce, "principal_id": principal, "host": "private.example", "action": "lifecycle_transition", "request": request})})
                return dict(connection.execute(text("SELECT public.tenant_control_lifecycle_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "request": __import__('json').dumps(request)}).scalar_one())

        assert transition("freeze", 0, "freeze-1")["state"] == "frozen"
        assert transition("begin_recovery", 1, "recover-1", approval="approval-1")["state"] == "recovering"
        assert transition("activate", 2, "activate-1", approval="approval-2")["state"] == "active"
        with admin.begin() as connection:
            connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            connection.execute(text("INSERT INTO public.tenant_legal_holds(hold_ref,tenant_id,state,evidence_digest) VALUES('hold-1',:tenant_id,'active',:digest)"), {"tenant_id": tenant_id, "digest": "a" * 64})
        with pytest.raises(SQLAlchemyError, match="tenant_hold_active"):
            transition("mark_deletion_candidate", 3, "delete-candidate-1", approval="approval-3", backup="b" * 64, retention="60")
        with admin.begin() as connection:
            connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            connection.execute(text("UPDATE public.tenant_legal_holds SET state='released',released_at=clock_timestamp() WHERE hold_ref='hold-1'"))
        candidate = transition("mark_deletion_candidate", 3, "delete-candidate-1", approval="approval-3", backup="b" * 64, retention="60")
        assert candidate["state"] == "deletion_candidate"
        assert transition("rollback", 4, "rollback-1")["state"] == "frozen"
        candidate_again = transition("mark_deletion_candidate", 5, "delete-candidate-2", approval="approval-4", backup="c" * 64, retention="120")
        assert candidate_again["state"] == "deletion_candidate"
        assert transition("delete", 6, "delete-1", approval="approval-5", backup="d" * 64, retention="120")["state"] == "deleted"
        # Deleted tenants cannot even receive a lifecycle claim, so rollback
        # is intentionally impossible after the destructive transition.
        with pytest.raises(SQLAlchemyError, match="tenant_membership_denied"):
            transition("rollback", 7, "rollback-after-delete")
    finally:
        if control is not None:
            control.dispose()
        with admin.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_legal_holds WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_lifecycle_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
        admin.dispose()


def test_wp10_live_tenant_data_blocks_downgrade_before_503_gateway_can_return():
    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    tenant_id = f"wp10-downgrade-{uuid.uuid4().hex[:12]}"
    engine = create_engine(database_url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
        result = subprocess.run(
            [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), "downgrade", "20260817_520"],
            cwd=ROOT,
            env={**os.environ, "MIGRATION_DATABASE_URL": database_url},
            capture_output=True,
            text=True,
        )
        assert result.returncode != 0
        assert "downgrade refused" in result.stderr and "tenant" in result.stderr
        with engine.connect() as connection:
            assert connection.execute(text("SELECT version_num FROM public.alembic_version")).scalar_one() == "20260817_525"
    finally:
        with engine.begin() as cleanup:
            cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
            cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
        engine.dispose()


def test_wp10_low_privilege_migrator_reaches_the_tenant_foundation_head(wp06_head):
    engine = create_engine(wp06_head.migration_url, pool_pre_ping=True)
    try:
        with engine.connect() as connection:
            head = connection.execute(text("SELECT version_num FROM public.alembic_version")).scalar_one()
            privileges = connection.execute(text("SELECT rolsuper,rolcreaterole FROM pg_roles WHERE rolname=current_user")).one()
            assert head == "20260817_525"
            assert tuple(privileges) == (False, False)
    finally:
        engine.dispose()