DataOps-platform/tests/integration/test_wp12_metering_showback_postgres.py

"""Real PostgreSQL gateway, concurrency, restart and downgrade checks for WP12."""

from __future__ import annotations

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

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

ROOT = Path(__file__).resolve().parents[2]


def _alembic(database_url: str, command: str, revision: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), command, revision],
        cwd=ROOT,
        env={**os.environ, "MIGRATION_DATABASE_URL": database_url},
        capture_output=True,
        text=True,
    )


def _claim(engine, action: str, principal: str) -> str:
    with engine.begin() as connection:
        connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_bi_ai_catalog_control"))
        return str(
            connection.execute(
                text("SELECT public.metering_showback_issue_claim(:action,:principal,'{}'::jsonb)"),
                {"action": action, "principal": principal},
            ).scalar_one()["request_claim"]
        )


def _runtime_write(engine, action: str, payload: dict) -> dict:
    with engine.begin() as connection:
        connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
        return dict(
            connection.execute(
                text("SELECT public.metering_showback_runtime_write(:action,CAST(:payload AS jsonb))"),
                {"action": action, "payload": json.dumps(payload, sort_keys=True)},
            ).scalar_one()
        )


def _control_write(engine, action: str, principal: str, payload: dict) -> dict:
    with engine.begin() as connection:
        connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_bi_ai_catalog_control"))
        return dict(
            connection.execute(
                text("SELECT public.metering_showback_control_write(:action,:principal,CAST(:payload AS jsonb))"),
                {"action": action, "principal": principal, "payload": json.dumps(payload, sort_keys=True)},
            ).scalar_one()
        )


@pytest.mark.integration
def test_wp12_upgrade_preflight_refuses_illegal_548_correction_fact():
    base_url = os.getenv("TEST_DATABASE_URL")
    if not base_url:
        pytest.skip("TEST_DATABASE_URL is required")
    base = make_url(base_url)
    database = f"wp12_preflight_{uuid.uuid4().hex}"
    admin = create_engine(base.set(database="postgres").render_as_string(hide_password=False), isolation_level="AUTOCOMMIT")
    quoted = admin.dialect.identifier_preparer.quote(database)
    try:
        with admin.connect() as connection:
            connection.execute(text(f"CREATE DATABASE {quoted}"))
        database_url = base.set(database=database).render_as_string(hide_password=False)
        with create_engine(database_url).begin() as connection:
            connection.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
            # Match the fixed-owner schema capability provided by role-init.
            connection.execute(text("GRANT USAGE,CREATE ON SCHEMA public TO dataops_tenant_foundation_owner"))
        assert _alembic(database_url, "stamp", "20260818_545").returncode == 0
        assert _alembic(database_url, "upgrade", "20260818_548").returncode == 0
        engine = create_engine(database_url)
        try:
            with engine.begin() as connection:
                values = {"uid": "wp12-preflight-original", "tenant": "local-engineering", "domain": "default", "correction": None}
                connection.execute(text("INSERT INTO public.metering_events(event_uid,tenant_ref,domain_ref,department_ref,project_ref,cost_center_ref,event_kind,occurred_at,window_start,window_end,quantity_micros,unit,idempotency_key,evidence_digest,evidence_reference,request_digest,correction_of,lease_fence) VALUES(:uid,:tenant,:domain,'engineering','local-engineering','local-fixture','query','2026-08-18T00:00:00Z','2026-08-18T00:00:00Z','2026-08-18T00:05:00Z',1,'requests',:uid,:digest,'local-fixture://wp12/v1',:digest,:correction,0)"), values | {"digest": "a" * 64})
                connection.execute(text("INSERT INTO public.metering_events(event_uid,tenant_ref,domain_ref,department_ref,project_ref,cost_center_ref,event_kind,occurred_at,window_start,window_end,quantity_micros,unit,idempotency_key,evidence_digest,evidence_reference,request_digest,correction_of,lease_fence) VALUES('wp12-preflight-illegal','local-engineering','other','engineering','local-engineering','local-fixture','query','2026-08-18T00:00:00Z','2026-08-18T00:00:00Z','2026-08-18T00:05:00Z',1,'requests','wp12-preflight-illegal',:digest,'local-fixture://wp12/v1',:digest,'wp12-preflight-original',0)"), {"digest": "b" * 64})
        finally:
            engine.dispose()
        assert _alembic(database_url, "upgrade", "20260818_549").returncode == 0
        rejected = _alembic(database_url, "upgrade", "20260818_550")
        assert rejected.returncode != 0 and "upgrade refused: WP12 integrity preflight failed" in (rejected.stderr + rejected.stdout)
    finally:
        with admin.connect() as connection:
            connection.execute(text("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=:database"), {"database": database})
            connection.execute(text(f"DROP DATABASE IF EXISTS {quoted}"))
        admin.dispose()


@pytest.mark.integration
def test_wp12_restricted_migrator_transfers_definer_owners_and_can_be_removed():
    base_url = os.getenv("TEST_DATABASE_URL")
    if not base_url:
        pytest.skip("TEST_DATABASE_URL is required")
    base = make_url(base_url)
    database = f"wp12_owner_{uuid.uuid4().hex}"
    migrator = f"wp12_migrator_{uuid.uuid4().hex[:12]}"
    password = f"migrator-{uuid.uuid4().hex}"
    admin = create_engine(base.set(database="postgres").render_as_string(hide_password=False), isolation_level="AUTOCOMMIT")
    quoted_database = admin.dialect.identifier_preparer.quote(database)
    quoted_migrator = admin.dialect.identifier_preparer.quote(migrator)
    try:
        with admin.connect() as connection:
            connection.execute(text(f"CREATE DATABASE {quoted_database}"))
        database_url = base.set(database=database).render_as_string(hide_password=False)
        with create_engine(database_url).begin() as connection:
            connection.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
        assert _alembic(database_url, "stamp", "20260818_545").returncode == 0
        assert _alembic(database_url, "upgrade", "20260818_551").returncode == 0
        with create_engine(database_url).begin() as connection:
            connection.execute(text(f"CREATE ROLE {quoted_migrator} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": password})
            connection.execute(text(f"GRANT dataops_tenant_foundation_owner TO {quoted_migrator}"))
            connection.execute(text(f"GRANT USAGE ON SCHEMA public TO {quoted_migrator}"))
            connection.execute(text("GRANT USAGE,CREATE ON SCHEMA public TO dataops_tenant_foundation_owner"))
            connection.execute(text(f"GRANT SELECT,UPDATE ON public.alembic_version TO {quoted_migrator}"))
            # 551 was created by the temporary migration login in the defect
            # scenario.  Simulate that exact owner state before the restricted
            # login upgrades to 552; it may only transfer functions it owns.
            functions = connection.execute(text("SELECT p.proname, pg_get_function_identity_arguments(p.oid) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace WHERE n.nspname='public' AND p.prosecdef AND p.proname LIKE 'metering_showback_%'" )).all()
            for name, args in functions:
                quoted_name = connection.dialect.identifier_preparer.quote(name)
                connection.execute(text(f"ALTER FUNCTION public.{quoted_name}({args}) OWNER TO {quoted_migrator}"))
        restricted_url = base.set(database=database, username=migrator, password=password).render_as_string(hide_password=False)
        restricted_upgrade = _alembic(restricted_url, "upgrade", "20260818_552")
        assert restricted_upgrade.returncode == 0, restricted_upgrade.stderr + restricted_upgrade.stdout
        with create_engine(database_url).connect() as connection:
            assert connection.execute(text("SELECT bool_and(r.rolname='dataops_tenant_foundation_owner' AND NOT r.rolcanlogin) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace JOIN pg_roles r ON r.oid=p.proowner WHERE n.nspname='public' AND p.prosecdef AND p.proname LIKE 'metering_showback_%'" )).scalar_one()
        assert _alembic(restricted_url, "downgrade", "20260818_551").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_552").returncode == 0
    finally:
        with admin.connect() as connection:
            connection.execute(text("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=:database"), {"database": database})
            connection.execute(text(f"DROP DATABASE IF EXISTS {quoted_database}"))
            connection.execute(text(f"REVOKE dataops_tenant_foundation_owner FROM {quoted_migrator}")) if connection.execute(text("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname=:role)"), {"role": migrator}).scalar_one() else None
            connection.execute(text(f"DROP ROLE IF EXISTS {quoted_migrator}"))
        admin.dispose()


@pytest.mark.integration
def test_wp12_real_postgres_migration_acl_replay_fence_restart_and_downgrade(monkeypatch):
    base_url = os.getenv("TEST_DATABASE_URL")
    if not base_url:
        pytest.skip("TEST_DATABASE_URL is required")
    base = make_url(base_url)
    database = f"wp12_metering_{uuid.uuid4().hex}"
    admin = create_engine(base.set(database="postgres").render_as_string(hide_password=False), isolation_level="AUTOCOMMIT")
    quoted = admin.dialect.identifier_preparer.quote(database)
    runtime_user, control_user = (f"wp12_api_runtime_{uuid.uuid4().hex[:12]}", f"wp12_api_control_{uuid.uuid4().hex[:12]}")
    runtime_password, control_password = (f"runtime-{uuid.uuid4().hex}", f"control-{uuid.uuid4().hex}")
    try:
        with admin.connect() as connection:
            connection.execute(text(f"CREATE DATABASE {quoted}"))
        database_url = base.set(database=database).render_as_string(hide_password=False)
        with create_engine(database_url).begin() as connection:
            connection.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
            # Match the fixed-owner schema capability provided by role-init.
            connection.execute(text("GRANT USAGE,CREATE ON SCHEMA public TO dataops_tenant_foundation_owner"))
        assert _alembic(database_url, "stamp", "20260818_545").returncode == 0
        upgraded = _alembic(database_url, "upgrade", "20260818_552")
        assert upgraded.returncode == 0, upgraded.stderr
        engine = create_engine(database_url, pool_pre_ping=True)
        try:
            with engine.begin() as connection:
                quote = connection.dialect.identifier_preparer.quote
                connection.execute(text(f"CREATE ROLE {quote(runtime_user)} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": runtime_password})
                connection.execute(text(f"CREATE ROLE {quote(control_user)} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION INHERIT PASSWORD :password"), {"password": control_password})
                connection.execute(text(f"GRANT dataops_app_runtime TO {quote(runtime_user)}"))
                connection.execute(text(f"GRANT dataops_bi_ai_catalog_issuer TO {quote(control_user)}"))
                connection.execute(text(f"GRANT EXECUTE ON FUNCTION public.metering_showback_issue_claim(text,text,jsonb),public.metering_showback_control_write(text,text,jsonb) TO {quote(control_user)}"))
                connection.execute(text("INSERT INTO public.metering_scope_grants(principal_ref,tenant_ref,domain_ref,role_name) VALUES('wp12-operator','local-engineering','default','operator'),('01900000-0000-7000-8000-000000068912','local-engineering','default','admin'),('wp12-other-domain','local-engineering','other','operator')"))
                assert not connection.execute(text("SELECT has_table_privilege('dataops_app_runtime','public.metering_events','SELECT,INSERT,UPDATE,DELETE,TRUNCATE')")).scalar_one()
                assert connection.execute(text("SELECT bool_and(r.rolname='dataops_tenant_foundation_owner' AND NOT r.rolcanlogin) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace JOIN pg_roles r ON r.oid=p.proowner WHERE n.nspname='public' AND p.prosecdef AND p.proname LIKE 'metering_showback_%'" )).scalar_one()
            lease_a = _runtime_write(engine, "claim_lease", {"request_claim": _claim(engine, "lease", "wp12-operator"), "lease_owner": "worker-a", "lease_token": str(uuid.uuid4())})
            with pytest.raises(Exception, match="metering_lease_unavailable"):
                _runtime_write(engine, "claim_lease", {"request_claim": _claim(engine, "lease", "wp12-operator"), "lease_owner": "worker-b", "lease_token": str(uuid.uuid4())})
            with engine.connect() as connection:
                lease_token = connection.execute(text("SELECT lease_token::text FROM public.metering_runtime_leases WHERE tenant_ref='local-engineering'")).scalar_one()
            event = {"request_claim": _claim(engine, "record", "wp12-operator"), "lease_owner": "worker-a", "lease_token": lease_token, "lease_fence": lease_a["lease_fence"], "event_uid": "wp12-event-001", "event_kind": "query", "occurred_at": "2026-08-18T00:00:00Z", "window_start": "2026-08-18T00:00:00Z", "window_end": "2026-08-18T00:05:00Z", "quantity_micros": 1250000, "unit": "gb", "idempotency_key": "wp12-idempotency-001", "evidence": {"digest": "a" * 64, "reference": "local-fixture://wp12/v1"}, "mapping": {"department": "engineering", "business_domain": "default", "project": "local-engineering", "cost_center": "local-fixture"}}
            assert _runtime_write(engine, "record_event", event)["replay"] is False
            replay = event | {"request_claim": _claim(engine, "record", "wp12-operator")}
            assert _runtime_write(engine, "record_event", replay)["replay"] is True
            with engine.begin() as connection:
                connection.execute(text("UPDATE public.metering_runtime_leases SET lease_expires_at=clock_timestamp()-interval '1 second'"))
            monkeypatch.setenv("DATABASE_URL", base.set(database=database, username=runtime_user, password=runtime_password).render_as_string(hide_password=False))
            monkeypatch.setenv("METERING_SHOWBACK_CONTROL_DATABASE_URL", base.set(database=database, username=control_user, password=control_password).render_as_string(hide_password=False))
            monkeypatch.setenv("TRUSTED_METERING_DOMAIN", "default")
            monkeypatch.setattr("app.core.system.auth.load_identity_from_token", lambda token, secret: {"id": "01900000-0000-7000-8000-000000068912", "roles": ["admin"]} if token == "wp12-admin" else None)
            from app import create_app

            app = create_app()
            app.config.update(TESTING=True)
            client = app.test_client()
            flask_event = {
                "schema_version": 1,
                "event_uid": "wp12-flask-event-001",
                "event_kind": "query",
                "occurred_at": "2026-08-18T00:10:00Z",
                "window_start": "2026-08-18T00:10:00Z",
                "window_end": "2026-08-18T00:15:00Z",
                "quantity": "0.500000",
                "unit": "gb",
                "idempotency_key": "wp12-flask-idempotency-001",
                "evidence": {"digest": "c" * 64, "reference": "local-fixture://wp12/v1"},
                "mapping": {"department": "engineering", "project": "local-engineering", "cost_center": "local-fixture"},
            }
            created_event = client.post("/api/system/metering/events", json=flask_event, headers={"Authorization": "Bearer wp12-admin"})
            assert created_event.status_code == 201 and created_event.headers["Cache-Control"] == "no-store"
            # The client cannot claim a domain; its three-field mapping is completed only by the trusted server value.
            asserted_invalid = client.post("/api/system/metering/events", json=flask_event | {"event_uid": "wp12-flask-event-invalid", "idempotency_key": "wp12-flask-invalid", "mapping": flask_event["mapping"] | {"business_domain": "other"}}, headers={"Authorization": "Bearer wp12-admin"})
            assert asserted_invalid.status_code == 400 and asserted_invalid.headers["Cache-Control"] == "no-store"
            with engine.begin() as connection:
                connection.execute(text("UPDATE public.metering_runtime_leases SET lease_expires_at=clock_timestamp()-interval '1 second'"))
            changed_event = client.post("/api/system/metering/events", json=flask_event | {"quantity": "0.750000"}, headers={"Authorization": "Bearer wp12-admin"})
            assert changed_event.status_code == 400 and changed_event.headers["Cache-Control"] == "no-store"
            # A gateway failure must roll back and leave the next HTTP write usable.
            resumed_event = client.post("/api/system/metering/events", json=flask_event | {"event_uid": "wp12-flask-event-002", "idempotency_key": "wp12-flask-idempotency-002", "quantity": "0.250000", "evidence": {"digest": "d" * 64, "reference": "local-fixture://wp12/v1"}}, headers={"Authorization": "Bearer wp12-admin"})
            assert resumed_event.status_code == 201 and resumed_event.headers["Cache-Control"] == "no-store"
            with engine.connect() as connection:
                assert connection.execute(text("SELECT count(*) FROM public.metering_events WHERE event_uid IN ('wp12-flask-event-001','wp12-flask-event-002') AND domain_ref='default'" )).scalar_one() == 2
            flask_allocation = {"schema_version": 1, "rule_uid": "wp12-flask-allocation-001", "rule_version": 1, "effective_start": "2026-08-01T00:00:00Z", "effective_end": "2026-09-01T00:00:00Z", "mapping": {"department": "engineering", "project": "local-engineering", "cost_center": "local-fixture"}, "allocations": [{"target": "local-engineering", "weight_micros": 1000000}]}
            created = client.post("/api/system/metering/allocations", json=flask_allocation, headers={"Authorization": "Bearer wp12-admin"})
            assert created.status_code == 201 and created.headers["Cache-Control"] == "no-store"
            changed_allocation = client.post("/api/system/metering/allocations", json=flask_allocation | {"allocations": [{"target": "local-engineering", "weight_micros": 999999}, {"target": "shared-services", "weight_micros": 1}]}, headers={"Authorization": "Bearer wp12-admin"})
            assert changed_allocation.status_code == 400 and changed_allocation.headers["Cache-Control"] == "no-store"
            flask_budget = {"schema_version": 1, "budget_uid": "wp12-flask-budget-001", "window": "2026-08", "mapping": flask_allocation["mapping"], "limit_micros": 2000000, "threshold_micros": 1500000}
            persisted_budget = client.post("/api/system/metering/budgets", json=flask_budget, headers={"Authorization": "Bearer wp12-admin"})
            assert persisted_budget.status_code == 201 and persisted_budget.headers["Cache-Control"] == "no-store"
            invalid_budget = client.post("/api/system/metering/budgets", json=flask_budget | {"threshold_micros": 2000001}, headers={"Authorization": "Bearer wp12-admin"})
            assert invalid_budget.status_code == 400 and invalid_budget.headers["Cache-Control"] == "no-store"
            def concurrent_allocation(payload):
                allocation_barrier.wait(timeout=5)
                try:
                    return "ok", _control_write(engine, "allocation", "wp12-operator", payload)
                except Exception as exc:  # the database is the assertion boundary
                    return "error", str(exc)

            atomic_mapping = {"department": "atomic-engineering", "business_domain": "default", "project": "local-engineering", "cost_center": "local-fixture"}
            atomic_base = {"rule_uid": "wp12-atomic-001", "rule_version": "1", "effective_start": "2026-08-01T00:00:00.000000Z", "effective_end": "2026-09-01T00:00:00.000000Z", "mapping": atomic_mapping}
            atomic_payloads = [atomic_base | {"allocations": [{"target": "atomic-a", "weight_micros": 1000000}]}, atomic_base | {"allocations": [{"target": "atomic-b", "weight_micros": 1000000}]}]
            allocation_barrier = Barrier(2)
            with ThreadPoolExecutor(max_workers=2) as executor:
                atomic_outcomes = list(executor.map(concurrent_allocation, atomic_payloads))
            assert sorted(outcome[0] for outcome in atomic_outcomes) == ["error", "ok"]
            assert any("metering_allocation_conflict" in outcome[1] for outcome in atomic_outcomes if outcome[0] == "error")
            winning_payload = next(payload for payload, outcome in zip(atomic_payloads, atomic_outcomes, strict=True) if outcome[0] == "ok")
            assert _control_write(engine, "allocation", "wp12-operator", winning_payload)["replay"] is True
            with engine.connect() as connection:
                assert connection.execute(text("SELECT count(*),sum(weight_micros) FROM public.metering_allocations WHERE rule_uid='wp12-atomic-001' AND rule_version=1")).one() == (1, 1000000)
            overlap_mapping = atomic_mapping | {"department": "overlap-engineering"}
            overlap_base = {"rule_version": "1", "effective_start": "2026-08-01T00:00:00.000000Z", "effective_end": "2026-09-01T00:00:00.000000Z", "mapping": overlap_mapping, "allocations": [{"target": "overlap-target", "weight_micros": 1000000}]}
            allocation_barrier = Barrier(2)
            with ThreadPoolExecutor(max_workers=2) as executor:
                overlap_outcomes = list(executor.map(concurrent_allocation, [overlap_base | {"rule_uid": "wp12-overlap-atomic-a"}, overlap_base | {"rule_uid": "wp12-overlap-atomic-b"}]))
            assert sorted(outcome[0] for outcome in overlap_outcomes) == ["error", "ok"]
            assert any("metering_allocation_window_overlap" in outcome[1] or "metering_allocation_rules_scope_window_excl" in outcome[1] for outcome in overlap_outcomes if outcome[0] == "error")
            # A newly constructed Flask app observes the committed gateway facts.
            second_app = create_app()
            second_app.config.update(TESTING=True)
            second_client = second_app.test_client()
            shown = second_client.get("/api/system/metering/reconciliation?window=2026-08", headers={"Authorization": "Bearer wp12-admin"})
            assert shown.status_code == 200 and shown.headers["Cache-Control"] == "no-store"
            replayed_rule = client.get("/api/system/metering/allocation-replay?window=2026-08&rule_uid=wp12-flask-allocation-001&rule_version=1", headers={"Authorization": "Bearer wp12-admin"})
            assert replayed_rule.status_code == 200 and replayed_rule.headers["Cache-Control"] == "no-store"
            assert replayed_rule.get_json()["data"]["difference_micros"] == 0
            allocation = {"rule_uid": "wp12-flask-allocation-001", "rule_version": "1", "effective_start": "2026-08-01T00:00:00.000000Z", "effective_end": "2026-09-01T00:00:00.000000Z", "mapping": {"department": "engineering", "business_domain": "default", "project": "local-engineering", "cost_center": "local-fixture"}, "allocations": [{"target": "local-engineering", "weight_micros": 1000000}]}
            assert _control_write(engine, "allocation", "wp12-operator", allocation)["persisted_before_ack"] is True
            with pytest.raises(Exception, match="metering_allocation_target_duplicate"):
                _control_write(engine, "allocation", "wp12-operator", allocation | {"allocations": [{"target": "local-engineering", "weight_micros": 500000}, {"target": "local-engineering", "weight_micros": 500000}]})
            with pytest.raises(Exception, match="metering_allocation_weight_invalid"):
                _control_write(engine, "allocation", "wp12-operator", allocation | {"allocations": [{"target": "local-engineering", "weight_micros": 0}, {"target": "shared-services", "weight_micros": 1000000}]})
            with pytest.raises(Exception, match="metering_allocation_window_overlap"):
                _control_write(engine, "allocation", "wp12-operator", allocation | {"rule_uid": "wp12-overlap-001"})
            expired_rule = allocation | {"rule_uid": "wp12-expired-001", "mapping": allocation["mapping"] | {"department": "expired-engineering"}, "effective_end": "2026-08-15T00:00:00.000000Z"}
            assert _control_write(engine, "allocation", "wp12-operator", expired_rule)["persisted_before_ack"] is True
            with engine.begin() as connection:
                connection.execute(text("UPDATE public.metering_runtime_leases SET lease_expires_at=clock_timestamp()-interval '1 second'"))
            reset_token = str(uuid.uuid4())
            reset_lease = _runtime_write(engine, "claim_lease", {"request_claim": _claim(engine, "lease", "wp12-operator"), "lease_owner": "worker-a", "lease_token": reset_token})
            event = event | {"lease_token": reset_token, "lease_fence": reset_lease["lease_fence"]}
            expired_event = event | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": "wp12-event-expired-rule", "idempotency_key": "wp12-idempotency-expired-rule", "quantity_micros": 1000000, "mapping": event["mapping"] | {"department": "expired-engineering"}}
            assert _runtime_write(engine, "record_event", expired_event)["replay"] is False
            assert client.get("/api/system/metering/allocation-replay?window=2026-08&rule_uid=wp12-expired-001&rule_version=1", headers={"Authorization": "Bearer wp12-admin"}).status_code == 403
            multi_one = allocation | {"rule_uid": "wp12-multi-001", "mapping": allocation["mapping"] | {"department": "multi-engineering"}, "effective_end": "2026-08-15T00:00:00.000000Z"}
            multi_two = multi_one | {"rule_uid": "wp12-multi-002", "effective_start": "2026-08-15T00:00:00.000000Z", "effective_end": "2026-09-01T00:00:00.000000Z"}
            assert _control_write(engine, "allocation", "wp12-operator", multi_one)["persisted_before_ack"] is True
            assert _control_write(engine, "allocation", "wp12-operator", multi_two)["persisted_before_ack"] is True
            for suffix, when in (("one", "2026-08-10T00:00:00Z"), ("two", "2026-08-18T00:00:00Z")):
                assert _runtime_write(engine, "record_event", event | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": f"wp12-event-multi-{suffix}", "idempotency_key": f"wp12-idempotency-multi-{suffix}", "quantity_micros": 1000000, "occurred_at": when, "window_start": when, "window_end": when.replace("00:00:00Z", "00:05:00Z"), "mapping": event["mapping"] | {"department": "multi-engineering"}})["replay"] is False
            assert client.get("/api/system/metering/allocation-replay?window=2026-08&rule_uid=wp12-multi-001&rule_version=1", headers={"Authorization": "Bearer wp12-admin"}).status_code == 403
            with pytest.raises(Exception, match="metering_allocation_conflict"):
                _control_write(engine, "allocation", "wp12-operator", allocation | {"allocations": [{"target": "local-engineering", "weight_micros": 999999}, {"target": "shared-services", "weight_micros": 1}]})
            with pytest.raises(Exception, match="metering_allocation_closed"):
                _control_write(engine, "allocation", "wp12-operator", allocation | {"mapping": allocation["mapping"] | {"business_domain": "other"}})
            with pytest.raises(Exception, match="metering_evidence_reference_invalid"):
                _runtime_write(engine, "record_event", event | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": "wp12-event-bad-ref", "idempotency_key": "wp12-idempotency-bad-ref", "evidence": {"digest": "b" * 64, "reference": "local-fixture://wp12/v1/SELECT"}})
            with pytest.raises(Exception, match="metering_correction_scope_invalid"):
                _runtime_write(engine, "record_event", event | {"request_claim": _claim(engine, "record", "wp12-other-domain"), "event_uid": "wp12-event-cross-domain-correction", "idempotency_key": "wp12-idempotency-cross-domain-correction", "mapping": event["mapping"] | {"business_domain": "other"}, "correction_of": "wp12-event-001"})
            valid_correction = event | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": "wp12-event-correction-001", "idempotency_key": "wp12-idempotency-correction-001", "quantity_micros": 0, "correction_of": "wp12-event-001"}
            assert _runtime_write(engine, "record_event", valid_correction)["replay"] is False
            with pytest.raises(Exception, match="metering_correction_scope_invalid"):
                _runtime_write(engine, "record_event", valid_correction | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": "wp12-event-correction-cycle", "idempotency_key": "wp12-idempotency-correction-cycle", "correction_of": "wp12-event-correction-001"})
            budget = {"budget_uid": "wp12-budget-001", "window": "2026-08", "mapping": allocation["mapping"], "limit_micros": "2000000", "threshold_micros": "1000000"}
            with ThreadPoolExecutor(max_workers=2) as executor:
                outcomes = list(executor.map(lambda _: _control_write(engine, "budget", "wp12-operator", budget)["alert_created"], range(2)))
            assert sorted(outcomes) == [False, True]
            with engine.begin() as connection:
                assert connection.execute(text("SELECT count(*) FROM public.metering_alert_outbox")).scalar_one() == 2
                connection.execute(text("UPDATE public.metering_runtime_leases SET lease_expires_at=clock_timestamp()-interval '1 second'"))
            lease_b = _runtime_write(engine, "claim_lease", {"request_claim": _claim(engine, "lease", "wp12-operator"), "lease_owner": "worker-b", "lease_token": str(uuid.uuid4())})
            with pytest.raises(Exception, match="metering_stale_fence"):
                _runtime_write(engine, "record_event", event | {"request_claim": _claim(engine, "record", "wp12-operator"), "event_uid": "wp12-event-stale", "idempotency_key": "wp12-idempotency-stale"})
            assert lease_b["lease_fence"] > lease_a["lease_fence"]
            for statement, error in (("SELECT count(*) FROM public.metering_events", "permission denied"), ("SELECT public.metering_showback_control_write('budget','wp12-operator','{}'::jsonb)", "permission denied"), ("SELECT public.metering_showback_runtime_write('chargeback','{}'::jsonb)", "chargeback_disabled")):
                with engine.begin() as connection:
                    connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
                    with pytest.raises(Exception, match=error):
                        connection.execute(text(statement))
            with engine.begin() as connection:
                connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
                report = connection.execute(text("SELECT public.metering_showback_rollup(CAST(:payload AS jsonb))"), {"payload": json.dumps({"request_claim": _claim(engine, "read", "wp12-operator"), "window": "2026-08"})}).scalar_one()
                assert report["source_micros"] == 5000000 and report["allocated_micros"] == 4000000 and report["variance_micros"] == 1000000
                assert report["allocation_rules"][0]["rule_uid"] == "wp12-flask-allocation-001"
                assert not connection.execute(text("SELECT has_function_privilege('dataops_app_runtime','public.metering_showback_runtime_read_legacy(text,jsonb)','EXECUTE')")).scalar_one()
                with pytest.raises(Exception, match="permission denied"):
                    connection.execute(text("SELECT public.metering_showback_runtime_read_legacy('showback','{}'::jsonb)"))
        finally:
            engine.dispose()
        downgraded = _alembic(database_url, "downgrade", "20260818_545")
        assert downgraded.returncode != 0
        assert "downgrade refused: WP12 allocation facts are nonempty" in (downgraded.stderr + downgraded.stdout)
    finally:
        with admin.connect() as connection:
            connection.execute(text("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=:database"), {"database": database})
            connection.execute(text(f"DROP DATABASE IF EXISTS {quoted}"))
            connection.execute(text(f"REVOKE dataops_app_runtime FROM {runtime_user}")) if connection.execute(text("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname=:role)"), {"role": runtime_user}).scalar_one() else None
            connection.execute(text(f"REVOKE dataops_bi_ai_catalog_issuer FROM {control_user}")) if connection.execute(text("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname=:role)"), {"role": control_user}).scalar_one() else None
            connection.execute(text(f"DROP ROLE IF EXISTS {runtime_user}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {control_user}"))
        admin.dispose()


@pytest.mark.integration
def test_wp12_shared_head_roundtrip_and_role_init_replay(monkeypatch):
    """Exercise the actual local test DB only when it is exactly at 545/empty."""

    database_url = os.getenv("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    engine = create_engine(database_url, pool_pre_ping=True)
    base = make_url(database_url)
    runtime_user = f"wp12_runtime_{uuid.uuid4().hex[:12]}"
    control_user = f"wp12_control_{uuid.uuid4().hex[:12]}"
    runtime_password = f"runtime-{uuid.uuid4().hex}"
    control_password = f"control-{uuid.uuid4().hex}"
    try:
        with engine.connect() as connection:
            shared_head = connection.execute(text("SELECT version_num FROM public.alembic_version")).scalar_one()
            assert shared_head in {"20260818_551", "20260818_552"}
            assert connection.execute(text("SELECT count(*) FROM public.metering_events")).scalar_one() == 0
        if shared_head == "20260818_552":
            normalized = _alembic(database_url, "downgrade", "20260818_551")
            assert normalized.returncode == 0, normalized.stderr
        from app.core.edge_gateway.runtime_roles import provision_runtime_login

        monkeypatch.setenv("DB_ROLE_INIT_DATABASE_URL", database_url)
        monkeypatch.setenv("MIGRATION_DATABASE_URL", database_url)
        monkeypatch.setenv("DATABASE_URL", base.set(username=runtime_user, password=runtime_password).render_as_string(hide_password=False))
        monkeypatch.setenv("DATAOPS_RUNTIME_USER", runtime_user)
        monkeypatch.setenv("DATAOPS_RUNTIME_PASSWORD", runtime_password)
        monkeypatch.setenv("DATAOPS_MIGRATOR_USER", str(base.username))
        monkeypatch.setenv("BI_AI_CATALOG_CONTROL_DATABASE_URL", base.set(username=control_user, password=control_password).render_as_string(hide_password=False))
        monkeypatch.setenv("DATAOPS_BI_AI_CATALOG_CONTROL_USER", control_user)
        monkeypatch.setenv("DATAOPS_BI_AI_CATALOG_CONTROL_PASSWORD", control_password)
        # Role-init precedes the restricted ownership transfer, then runs a
        # second time below to prove replay cannot reopen the fenced gateways.
        provision_runtime_login()
        upgraded = _alembic(database_url, "upgrade", "20260818_552")
        assert upgraded.returncode == 0, upgraded.stderr
        provision_runtime_login()

        with engine.connect() as connection:
            assert not connection.execute(
                text("SELECT has_table_privilege(:runtime,'public.metering_events','SELECT,INSERT,UPDATE,DELETE,TRUNCATE')"),
                {"runtime": runtime_user},
            ).scalar_one()
            assert not connection.execute(
                text("SELECT has_function_privilege(:runtime,'public.metering_showback_runtime_read_legacy(text,jsonb)','EXECUTE')"),
                {"runtime": runtime_user},
            ).scalar_one()
            assert connection.execute(
                text("SELECT bool_and(r.rolname='dataops_tenant_foundation_owner' AND NOT r.rolcanlogin) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace JOIN pg_roles r ON r.oid=p.proowner WHERE n.nspname='public' AND p.prosecdef AND p.proname LIKE 'metering_showback_%'")
            ).scalar_one()
            assert connection.execute(
                text("SELECT has_function_privilege(:control,'public.metering_showback_issue_claim(text,text,jsonb)','EXECUTE')"),
                {"control": control_user},
            ).scalar_one()
            assert connection.execute(
                text("SELECT has_function_privilege(:control,'public.metering_showback_control_write(text,text,jsonb)','EXECUTE')"),
                {"control": control_user},
            ).scalar_one()
            assert connection.execute(
                text(
                    "SELECT count(*)=0 FROM ("
                    "SELECT count(*) AS n FROM public.metering_scope_grants UNION ALL SELECT count(*) FROM public.metering_runtime_leases "
                    "UNION ALL SELECT count(*) FROM public.metering_runtime_claims UNION ALL SELECT count(*) FROM public.metering_events "
                    "UNION ALL SELECT count(*) FROM public.metering_allocation_rules UNION ALL SELECT count(*) FROM public.metering_allocations "
                    "UNION ALL SELECT count(*) FROM public.metering_budgets UNION ALL SELECT count(*) FROM public.metering_alert_outbox "
                    "UNION ALL SELECT count(*) FROM public.metering_reconciliation_reports UNION ALL SELECT count(*) FROM public.metering_audit_events"
                    ") facts WHERE n<>0"
                )
            ).scalar_one()
        downgraded = _alembic(database_url, "downgrade", "20260818_551")
        assert downgraded.returncode == 0, downgraded.stderr
        replayed = _alembic(database_url, "upgrade", "20260818_552")
        assert replayed.returncode == 0, replayed.stderr
    finally:
        # The test has no WP12 facts; leave the shared local database at 552
        # and remove its short-lived login identities without printing secrets.
        with engine.begin() as connection:
            connection.execute(
                text(
                        "DO $$ BEGIN IF EXISTS(SELECT 1 FROM pg_roles WHERE rolname=:control) THEN "
                            "IF to_regprocedure('public.metering_showback_issue_claim(text,text,jsonb)') IS NOT NULL THEN "
                        "EXECUTE 'REVOKE ALL ON FUNCTION public.metering_showback_issue_claim(text,text,jsonb) FROM ' || quote_ident(:control); END IF; "
                    "IF to_regprocedure('public.metering_showback_control_write(text,text,jsonb)') IS NOT NULL THEN "
                    "EXECUTE 'REVOKE ALL ON FUNCTION public.metering_showback_control_write(text,text,jsonb) FROM ' || quote_ident(:control); END IF; "
                        "IF to_regprocedure('public.bi_ai_catalog_issue_request_claim(text,text,jsonb)') IS NOT NULL THEN "
                        "EXECUTE 'REVOKE ALL ON FUNCTION public.bi_ai_catalog_issue_request_claim(text,text,jsonb) FROM ' || quote_ident(:control); END IF; "
                        "END IF; END $$"
                ),
                {"control": control_user},
            )
            connection.execute(text(f"REVOKE dataops_app_runtime,dataops_agent_runtime FROM {runtime_user}"))
            connection.execute(text(f"REVOKE dataops_bi_ai_catalog_issuer FROM {control_user}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {runtime_user}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {control_user}"))
        engine.dispose()