DataOps-platform/tests/integration/test_production_operations_postgres.py

from __future__ import annotations

import os
import subprocess
import threading
import uuid
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.exc import DBAPIError, IntegrityError

pytestmark = pytest.mark.integration


DATABASE_URL = os.environ.get("TEST_DATABASE_URL")
ROOT = Path(__file__).resolve().parents[2]


def _database_url():
    if not DATABASE_URL:
        pytest.skip("TEST_DATABASE_URL is required")
    return DATABASE_URL


def _alembic(command: str, target: str) -> None:
    environment = dict(os.environ)
    environment["MIGRATION_DATABASE_URL"] = _database_url()
    subprocess.run(
        [
            str(ROOT / ".venv/bin/alembic"),
            "-c",
            str(ROOT / "alembic.ini"),
            command,
            target,
        ],
        cwd=ROOT,
        env=environment,
        check=True,
        capture_output=True,
        text=True,
    )


def _seed(connection, suffix: str):
    actor = str(uuid.uuid5(uuid.NAMESPACE_URL, f"wp05-test-actor-{suffix}"))
    incident = str(uuid.uuid5(uuid.NAMESPACE_URL, f"wp05-test-incident-{suffix}"))
    alert = str(uuid.uuid5(uuid.NAMESPACE_URL, f"wp05-test-alert-{suffix}"))
    username = f"wp05-test-{suffix}"[:80]
    connection.execute(
        text(
            "INSERT INTO public.users "
            "(id,username,display_name,password_hash,status) "
            "VALUES (CAST(:id AS uuid),:username,:username,'test','active')"
        ),
        {"id": actor, "username": username},
    )
    connection.execute(
        text(
            "INSERT INTO public.data_incidents "
            "(uid,code,dedup_key,title,severity,status,owner_uid,escalation_level,"
            "first_detected_at,last_observed_at,created_by) VALUES "
            "(CAST(:uid AS uuid),:code,:digest,'wp05 test','critical','open',"
            "CAST(:actor AS uuid),1,clock_timestamp(),clock_timestamp(),"
            "CAST(:actor AS uuid))"
        ),
        {
            "uid": incident,
            "code": f"WP05-TEST-{suffix}",
            "digest": uuid.uuid5(uuid.NAMESPACE_URL, f"incident-{suffix}").hex,
            "actor": actor,
        },
    )
    connection.execute(
        text(
            "INSERT INTO public.data_observability_alerts "
            "(uid,dedup_key,incident_uid,layer,sli_type,title,severity,status,"
            "occurrence_count,escalation_level,owner_uid,first_observed_at,"
            "last_observed_at,delivery_status,evidence) VALUES "
            "(CAST(:uid AS uuid),:digest,CAST(:incident AS uuid),'service',"
            "'delivery','wp05 test','critical','open',1,1,CAST(:actor AS uuid),"
            "clock_timestamp(),clock_timestamp(),'pending','{}')"
        ),
        {
            "uid": alert,
            "digest": uuid.uuid5(uuid.NAMESPACE_URL, f"alert-{suffix}").hex,
            "incident": incident,
            "actor": actor,
        },
    )
    return {"actor": actor, "incident": incident, "alert": alert}


def _cleanup_wp05_test_rows(engine) -> None:
    """Only remove this module's explicitly namespaced synthetic rows."""
    with engine.begin() as connection:
        # The production ledger is append-only even to its evidence owner.  The
        # integration harness is the privileged migration operator and disables
        # only this user trigger briefly to remove its own namespaced fixtures.
        connection.execute(text("SET LOCAL ROLE dataops_edge_evidence_owner"))
        connection.execute(
            text(
                "ALTER TABLE public.production_observability_audits "
                "DISABLE TRIGGER USER"
            )
        )
        connection.execute(text("RESET ROLE"))
        connection.execute(
            text(
                "DELETE FROM public.production_observability_audits a "
                "USING public.users u WHERE a.actor_uid=u.id "
                "AND u.username LIKE 'wp05-test-%'"
            )
        )
        connection.execute(
            text(
                "SET LOCAL ROLE dataops_edge_evidence_owner"
            )
        )
        connection.execute(
            text(
                "ALTER TABLE public.production_observability_audits "
                "ENABLE TRIGGER USER"
            )
        )
        connection.execute(text("RESET ROLE"))
        if connection.execute(
            text("SELECT to_regclass('public.production_observability_itsm_references')")
        ).scalar_one():
            connection.execute(
                text(
                    "DELETE FROM public.production_observability_itsm_references r "
                    "USING public.data_incidents i WHERE r.incident_uid=i.uid "
                    "AND i.code LIKE 'WP05-TEST-%'"
                )
            )
        connection.execute(
            text(
                "DELETE FROM public.production_observability_deliveries d "
                "USING public.data_incidents i WHERE d.incident_uid=i.uid "
                "AND i.code LIKE 'WP05-TEST-%'"
            )
        )
        connection.execute(
            text(
                "DELETE FROM public.production_observability_policies p "
                "USING public.users u WHERE p.created_by=u.id "
                "AND u.username LIKE 'wp05-test-%'"
            )
        )
        connection.execute(
            text(
                "DELETE FROM public.data_observability_alerts "
                "WHERE incident_uid IN (SELECT uid FROM public.data_incidents "
                "WHERE code LIKE 'WP05-TEST-%')"
            )
        )
        connection.execute(
            text("DELETE FROM public.data_incidents WHERE code LIKE 'WP05-TEST-%'")
        )
        connection.execute(
            text("DELETE FROM public.users WHERE username LIKE 'wp05-test-%'")
        )


@pytest.fixture(scope="module")
def pg_engine():
    _alembic("upgrade", "head")
    engine = create_engine(_database_url(), pool_pre_ping=True)
    _cleanup_wp05_test_rows(engine)
    yield engine
    _cleanup_wp05_test_rows(engine)
    engine.dispose()


def test_persistent_delivery_claim_fencing_backoff_and_policy_constraints():
    url = os.environ.get("TEST_DATABASE_URL")
    if not url:
        pytest.skip("TEST_DATABASE_URL is required")
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    engine = create_engine(url, pool_pre_ping=True)
    actor, incident, alert = (str(uuid.uuid4()) for _ in range(3))
    try:
        with engine.begin() as connection:
            connection.execute(
                text(
                    "INSERT INTO public.users (id,username,display_name,password_hash,status) VALUES (CAST(:id AS uuid),:username,:username,'test','active')"
                ),
                {"id": actor, "username": f"wp05ops{actor[:8]}"},
            )
            connection.execute(
                text(
                    "INSERT INTO public.data_incidents (uid,code,dedup_key,title,severity,status,owner_uid,escalation_level,first_detected_at,last_observed_at,created_by) VALUES (CAST(:uid AS uuid),:code,:digest,'synthetic','critical','open',CAST(:actor AS uuid),1,clock_timestamp(),clock_timestamp(),CAST(:actor AS uuid))"
                ),
                {
                    "uid": incident,
                    "code": f"INC-{incident[:8]}",
                    "digest": "a" * 64,
                    "actor": actor,
                },
            )
            connection.execute(
                text(
                    "INSERT INTO public.data_observability_alerts (uid,dedup_key,incident_uid,layer,sli_type,title,severity,status,occurrence_count,escalation_level,owner_uid,first_observed_at,last_observed_at,delivery_status,evidence) VALUES (CAST(:uid AS uuid),:digest,CAST(:incident AS uuid),'service','delivery','synthetic','critical','open',1,1,CAST(:actor AS uuid),clock_timestamp(),clock_timestamp(),'pending','{}')"
                ),
                {
                    "uid": alert,
                    "digest": "b" * 64,
                    "incident": incident,
                    "actor": actor,
                },
            )
            repository = SqlAlchemyProductionOperationsRepository(connection)
            service = ProductionOperationsService(repository, now=lambda: 100)
            delivery = service.enqueue_incident_delivery(
                {
                    "incident_uid": incident,
                    "alert_uid": alert,
                    "channel": "on_call",
                    "summary": "synthetic",
                },
                actor_uid=actor,
            )
            claim = service.claim_delivery(delivery["uid"], worker_id="worker-a")
            assert (
                service.record_delivery_attempt(
                    claim["uid"],
                    worker_id="worker-a",
                    lease_fence=claim["lease_fence"],
                    delivered=False,
                    actor_uid=actor,
                )["status"]
                == "pending"
            )
            with pytest.raises(ProductionOperationsError):
                service.claim_delivery(delivery["uid"], worker_id="worker-b")
            connection.execute(
                text(
                    "UPDATE public.production_observability_deliveries SET next_attempt_at=clock_timestamp() WHERE uid=CAST(:uid AS uuid)"
                ),
                {"uid": delivery["uid"]},
            )
            again = service.claim_delivery(delivery["uid"], worker_id="worker-b")
            assert again["lease_fence"] == 2
            with pytest.raises(IntegrityError):
                connection.execute(
                    text(
                        "INSERT INTO public.production_observability_policies (uid,code,version,layer,error_budget_ratio,status,created_by) VALUES (CAST(:uid AS uuid),'SVC','1.0.0','service',0.1,'active',CAST(:actor AS uuid)),(CAST(:uid2 AS uuid),'SVC','1.1.0','service',0.2,'active',CAST(:actor AS uuid))"
                    ),
                    {
                        "uid": str(uuid.uuid4()),
                        "uid2": str(uuid.uuid4()),
                        "actor": actor,
                    },
                )
    finally:
        with engine.begin() as connection:
            connection.execute(
                text(
                    "DELETE FROM public.production_observability_audits WHERE actor_uid=CAST(:actor AS uuid)"
                ),
                {"actor": actor},
            )
            connection.execute(
                text(
                    "DELETE FROM public.production_observability_deliveries WHERE incident_uid=CAST(:incident AS uuid)"
                ),
                {"incident": incident},
            )
            connection.execute(
                text(
                    "DELETE FROM public.production_observability_policies WHERE created_by=CAST(:actor AS uuid)"
                ),
                {"actor": actor},
            )
            connection.execute(
                text(
                    "DELETE FROM public.data_observability_alerts WHERE uid=CAST(:alert AS uuid)"
                ),
                {"alert": alert},
            )
            connection.execute(
                text(
                    "DELETE FROM public.data_incidents WHERE uid=CAST(:incident AS uuid)"
                ),
                {"incident": incident},
            )
            connection.execute(
                text("DELETE FROM public.users WHERE id=CAST(:actor AS uuid)"),
                {"actor": actor},
            )
        engine.dispose()


def test_failed_business_transaction_keeps_independent_rejection_audit():
    """RED: audit must not share the business transaction that is rolled back."""
    url = os.environ.get("TEST_DATABASE_URL")
    if not url:
        pytest.skip("TEST_DATABASE_URL is required")
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    audit_uid = str(uuid.uuid4())
    actor = str(uuid.uuid4())
    engine = create_engine(url, pool_pre_ping=True)
    try:
        with engine.begin() as connection:
            connection.execute(text("INSERT INTO public.users (id,username,display_name,password_hash,status) VALUES (CAST(:id AS uuid),:name,:name,'test','active')"), {"id": actor, "name": f"wp05audit{actor[:8]}"})
        with engine.connect() as connection:
            transaction = connection.begin()
            SqlAlchemyProductionOperationsRepository(connection).append_rejection_audit_independently({"uid": audit_uid, "actor_uid": actor, "action": "delivery_rejected", "safe_detail": {"reason_code": "synthetic"}})
            transaction.rollback()
        with engine.connect() as connection:
            assert connection.execute(text("SELECT count(*) FROM public.production_observability_audits WHERE uid=CAST(:uid AS uuid)"), {"uid": audit_uid}).scalar_one() == 1
    finally:
        with engine.begin() as connection:
            connection.execute(text("SET LOCAL ROLE dataops_edge_evidence_owner"))
            connection.execute(
                text(
                    "ALTER TABLE public.production_observability_audits "
                    "DISABLE TRIGGER USER"
                )
            )
            connection.execute(text("RESET ROLE"))
            connection.execute(text("DELETE FROM public.production_observability_audits WHERE uid=CAST(:uid AS uuid)"), {"uid": audit_uid})
            connection.execute(text("SET LOCAL ROLE dataops_edge_evidence_owner"))
            connection.execute(
                text(
                    "ALTER TABLE public.production_observability_audits "
                    "ENABLE TRIGGER USER"
                )
            )
            connection.execute(text("RESET ROLE"))
            connection.execute(text("DELETE FROM public.users WHERE id=CAST(:id AS uuid)"), {"id": actor})
        engine.dispose()


def test_two_postgresql_workers_barrier_claims_once_without_deadlock(pg_engine):
    """RED: database CAS must allow exactly one concurrent lease claimant."""
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "barrier-claim"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
            delivery = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: 1
            ).enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"],
                    "alert_uid": ids["alert"],
                    "channel": "on_call",
                    "summary": "wp05 test barrier claim",
                },
                actor_uid=ids["actor"],
            )

        barrier = threading.Barrier(2)
        outcomes = []
        failures = []
        lock = threading.Lock()

        def claim(worker_id):
            try:
                with pg_engine.begin() as connection:
                    service = ProductionOperationsService(
                        SqlAlchemyProductionOperationsRepository(connection), now=lambda: 1
                    )
                    barrier.wait(timeout=5)
                    try:
                        result = service.claim_delivery(
                            delivery["uid"], worker_id=worker_id
                        )
                    except ProductionOperationsError as error:
                        result = error
                with lock:
                    outcomes.append(result)
            except BaseException as error:  # assertions run on the parent thread
                with lock:
                    failures.append(error)

        workers = [
            threading.Thread(target=claim, args=("wp05-worker-a",)),
            threading.Thread(target=claim, args=("wp05-worker-b",)),
        ]
        for worker in workers:
            worker.start()
        for worker in workers:
            worker.join(timeout=10)

        assert not failures
        assert not any(worker.is_alive() for worker in workers)
        claimed = [item for item in outcomes if isinstance(item, dict)]
        rejected = [item for item in outcomes if isinstance(item, ProductionOperationsError)]
        assert len(claimed) == 1
        assert len(rejected) == 1
        assert claimed[0]["lease_fence"] == 1
    finally:
        _cleanup_wp05_test_rows(pg_engine)


def test_postgresql_db_time_rejects_expired_old_fence_before_reclaim(pg_engine):
    """RED: a late worker cannot write after DB lease expiry, even before reclaim."""
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "db-time-fence"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
            first = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: -10_000
            )
            delivery = first.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"],
                    "alert_uid": ids["alert"],
                    "channel": "smtp",
                    "summary": "wp05 test expiry",
                },
                actor_uid=ids["actor"],
            )
            claim = first.claim_delivery(delivery["uid"], worker_id="wp05-old-worker")
            connection.execute(
                text(
                    "UPDATE public.production_observability_deliveries "
                    "SET lease_expires_at=clock_timestamp()-interval '1 second' "
                    "WHERE uid=CAST(:uid AS uuid)"
                ),
                {"uid": delivery["uid"]},
            )

        with pg_engine.begin() as connection:
            old_worker = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: -99_999
            )
            for delivered in (True, False):
                with pytest.raises(ProductionOperationsError, match="expired or fenced"):
                    old_worker.record_delivery_attempt(
                        delivery["uid"],
                        worker_id="wp05-old-worker",
                        lease_fence=claim["lease_fence"],
                        delivered=delivered,
                        actor_uid=ids["actor"],
                    )

        with pg_engine.begin() as connection:
            recovered = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: 0
            )
            renewed = recovered.claim_delivery(
                delivery["uid"], worker_id="wp05-new-worker"
            )
            assert renewed["lease_fence"] == claim["lease_fence"] + 1
            assert (
                recovered.record_delivery_attempt(
                    delivery["uid"],
                    worker_id="wp05-new-worker",
                    lease_fence=renewed["lease_fence"],
                    delivered=True,
                    actor_uid=ids["actor"],
                )["status"]
                == "delivered"
            )
    finally:
        _cleanup_wp05_test_rows(pg_engine)


def test_postgresql_concurrent_enqueue_enforces_closed_hard_cap(pg_engine):
    """RED: concurrent distinct requests cannot exceed one configured queue slot."""
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "queue-cap"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
        barrier = threading.Barrier(2)
        results = []
        failures = []
        lock = threading.Lock()

        def enqueue(channel):
            try:
                with pg_engine.begin() as connection:
                    service = ProductionOperationsService(
                        SqlAlchemyProductionOperationsRepository(connection),
                        now=lambda: 0,
                        queue_hard_limit=1,
                    )
                    barrier.wait(timeout=5)
                    try:
                        result = service.enqueue_incident_delivery(
                            {
                                "incident_uid": ids["incident"],
                                "alert_uid": ids["alert"],
                                "channel": channel,
                                "summary": "wp05 test queue cap",
                            },
                            actor_uid=ids["actor"],
                        )
                    except ProductionOperationsError as error:
                        result = error
                with lock:
                    results.append(result)
            except BaseException as error:
                with lock:
                    failures.append(error)

        workers = [
            threading.Thread(target=enqueue, args=("monitoring",)),
            threading.Thread(target=enqueue, args=("on_call",)),
        ]
        for worker in workers:
            worker.start()
        for worker in workers:
            worker.join(timeout=10)
        assert not failures
        assert not any(worker.is_alive() for worker in workers)
        accepted = [item for item in results if isinstance(item, dict)]
        rejected = [item for item in results if isinstance(item, ProductionOperationsError)]
        assert len(accepted) == 1
        assert len(rejected) == 1
        assert "hard limit" in str(rejected[0])

        with pg_engine.begin() as connection:
            service = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection),
                now=lambda: 0,
                queue_hard_limit=1,
            )
            payload = {
                "incident_uid": ids["incident"],
                "alert_uid": ids["alert"],
                "channel": accepted[0]["channel"],
                "summary": "wp05 test queue cap",
            }
            assert service.enqueue_incident_delivery(payload, actor_uid=ids["actor"])["uid"] == accepted[0]["uid"]
            with pytest.raises(ProductionOperationsError, match="idempotency conflict"):
                service.enqueue_incident_delivery(
                    {**payload, "summary": "wp05 changed replay"}, actor_uid=ids["actor"]
                )
            assert connection.execute(
                text(
                    "SELECT count(*) FROM public.production_observability_deliveries "
                    "WHERE incident_uid=CAST(:incident AS uuid)"
                ),
                {"incident": ids["incident"]},
            ).scalar_one() == 1
    finally:
        _cleanup_wp05_test_rows(pg_engine)


def test_new_repository_instances_restore_operations_state_and_references(pg_engine):
    """RED: persisted rows, policies, and both ITSM directions survive a restart."""
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "restart-state"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
            first = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: 0
            )
            pending = first.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                    "channel": "monitoring", "summary": "wp05 pending",
                }, actor_uid=ids["actor"]
            )
            expired = first.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                    "channel": "smtp", "summary": "wp05 expired",
                }, actor_uid=ids["actor"]
            )
            first.claim_delivery(expired["uid"], worker_id="wp05-expired-worker")
            connection.execute(text(
                "UPDATE public.production_observability_deliveries "
                "SET lease_expires_at=clock_timestamp()-interval '1 second' "
                "WHERE uid=CAST(:uid AS uuid)"), {"uid": expired["uid"]})
            dead = first.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                    "channel": "on_call", "summary": "wp05 dead letter",
                }, actor_uid=ids["actor"]
            )
            connection.execute(text(
                "UPDATE public.production_observability_deliveries SET "
                "status='dead_letter',attempt_count=3,lease_owner=NULL,"
                "lease_expires_at=NULL WHERE uid=CAST(:uid AS uuid)"),
                {"uid": dead["uid"]})
            first.create_policy(
                {"code": "WP05_TEST", "version": "1.0.0", "layer": "service", "error_budget_ratio": 0.1},
                actor_uid=ids["actor"],
            )
            first.activate_policy("WP05_TEST", "1.0.0", actor_uid=ids["actor"])
            itsm = first.bind_itsm_reference(
                incident_uid=ids["incident"], external_ref="WP05-TEST-ITSM-1",
                idempotency_key="wp05-test-itsm-restart", actor_uid=ids["actor"],
            )

        with pg_engine.begin() as connection:
            restored = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: 0
            )
            assert {item["status"] for item in restored.list_deliveries(ids["incident"])} == {
                "pending", "processing", "dead_letter"
            }
            assert restored.active_policy("WP05_TEST")["version"] == "1.0.0"
            assert restored.itsm_reference_for_incident(ids["incident"]) == itsm
            assert restored.platform_incident_for_itsm("WP05-TEST-ITSM-1") == ids["incident"]
            assert restored.bind_itsm_reference(
                incident_uid=ids["incident"], external_ref="WP05-TEST-ITSM-1",
                idempotency_key="wp05-test-itsm-restart", actor_uid=ids["actor"],
            ) == itsm
            with pytest.raises(ProductionOperationsError, match="idempotency conflict"):
                restored.bind_itsm_reference(
                    incident_uid=ids["incident"], external_ref="WP05-TEST-ITSM-CHANGED",
                    idempotency_key="wp05-test-itsm-restart", actor_uid=ids["actor"],
                )
            renewed = restored.claim_delivery(expired["uid"], worker_id="wp05-restarted")
            assert renewed["lease_fence"] == 2
            assert restored.record_delivery_attempt(
                renewed["uid"], worker_id="wp05-restarted", lease_fence=renewed["lease_fence"],
                delivered=True, actor_uid=ids["actor"],
            )["status"] == "delivered"
            assert restored.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                    "channel": "monitoring", "summary": "wp05 pending",
                }, actor_uid=ids["actor"]
            )["uid"] == pending["uid"]
            with pytest.raises(ProductionOperationsError, match="idempotency conflict"):
                restored.enqueue_incident_delivery(
                    {
                        "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                        "channel": "monitoring", "summary": "wp05 changed pending",
                    }, actor_uid=ids["actor"]
                )
    finally:
        _cleanup_wp05_test_rows(pg_engine)


def test_480_migration_round_trip_and_constraints(pg_engine):
    """RED: 479↔480 is reversible when exact wp05 test data is absent."""
    _cleanup_wp05_test_rows(pg_engine)
    try:
        _alembic("downgrade", "20260809_479")
        with pg_engine.connect() as connection:
            assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == "20260809_479"
            assert connection.execute(text("SELECT to_regclass('public.production_observability_deliveries')")).scalar_one() is None
        _alembic("upgrade", "20260811_480")
        with pg_engine.connect() as connection:
            assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == "20260811_480"
            tables = connection.execute(text("""
                SELECT relname FROM pg_class JOIN pg_namespace ON pg_namespace.oid=pg_class.relnamespace
                WHERE nspname='public' AND relname LIKE 'production_observability_%' AND relkind='r'
            """)).scalars().all()
            assert {
                "production_observability_deliveries",
                "production_observability_audits",
                "production_observability_policies",
                "production_observability_itsm_references",
            } <= set(tables)
            constraints = connection.execute(text("""
                SELECT conname FROM pg_constraint
                WHERE conrelid='public.production_observability_deliveries'::regclass
            """)).scalars().all()
            assert "production_observability_deliveries_status_check" in constraints
            itsm_constraints = connection.execute(text("""
                SELECT conname FROM pg_constraint
                WHERE conrelid='public.production_observability_itsm_references'::regclass
            """)).scalars().all()
            assert {
                "production_observability_itsm_references_incident_uid_key",
                "production_observability_itsm_references_external_ref_key",
                "production_observability_itsm_references_idempotency_key_key",
            } <= set(itsm_constraints)
    finally:
        _alembic("upgrade", "head")


def test_runtime_cannot_mutate_append_only_audit_but_controlled_function_writes(pg_engine):
    """RED: runtime has SELECT+EXECUTE only; evidence remains after business rollback."""
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "audit-ledger"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
            audit_uid = str(uuid.uuid4())
            repository = SqlAlchemyProductionOperationsRepository(connection)
            repository.append_audit(
                {
                    "uid": audit_uid,
                    "delivery_uid": None,
                    "action": "delivery_enqueued",
                    "actor_uid": ids["actor"],
                    "safe_detail": {"channel": "on_call"},
                }
            )
            assert connection.execute(
                text("SELECT count(*) FROM public.production_observability_audits WHERE uid=CAST(:uid AS uuid)"),
                {"uid": audit_uid},
            ).scalar_one() == 1
        for statement in (
            "INSERT INTO public.production_observability_audits (uid,action,actor_uid,safe_detail) VALUES (gen_random_uuid(),'delivery_enqueued',CAST(:actor AS uuid),'{}')",
            "UPDATE public.production_observability_audits SET action='delivery_delivered'",
            "DELETE FROM public.production_observability_audits",
            "TRUNCATE public.production_observability_audits",
        ):
            with pg_engine.begin() as connection:
                connection.execute(text("SET LOCAL ROLE dataops_app_runtime"))
                with pytest.raises(DBAPIError):
                    connection.execute(text(statement), {"actor": ids["actor"]})
    finally:
        _cleanup_wp05_test_rows(pg_engine)


def test_persistent_dead_letter_compensation_is_atomic_and_exact(pg_engine):
    """RED: only dead_letter can compensate; same receipt replays, a changed one conflicts."""
    from app.core.events.production_operations import (
        ProductionOperationsError,
        ProductionOperationsService,
    )
    from app.core.events.production_operations_repository import (
        SqlAlchemyProductionOperationsRepository,
    )

    suffix = "compensation"
    _cleanup_wp05_test_rows(pg_engine)
    try:
        with pg_engine.begin() as connection:
            ids = _seed(connection, suffix)
            service = ProductionOperationsService(
                SqlAlchemyProductionOperationsRepository(connection), now=lambda: 1
            )
            delivery = service.enqueue_incident_delivery(
                {
                    "incident_uid": ids["incident"], "alert_uid": ids["alert"],
                    "channel": "on_call", "summary": "compensation test",
                },
                actor_uid=ids["actor"],
            )
            connection.execute(
                text("UPDATE public.production_observability_deliveries SET status='dead_letter' WHERE uid=CAST(:uid AS uuid)"),
                {"uid": delivery["uid"]},
            )
            compensated = service.compensate_dead_letter(
                delivery["uid"], reason_code="operator_review",
                receipt_code="OPS_REPLAY_1", actor_uid=ids["actor"],
            )
            assert compensated["status"] == "compensated"
            assert service.compensate_dead_letter(
                delivery["uid"], reason_code="operator_review",
                receipt_code="OPS_REPLAY_1", actor_uid=ids["actor"],
            )["uid"] == delivery["uid"]
            with pytest.raises(ProductionOperationsError, match="compensation conflict"):
                service.compensate_dead_letter(
                    delivery["uid"], reason_code="provider_recovered",
                    receipt_code="OPS_REPLAY_2", actor_uid=ids["actor"],
                )
    finally:
        _cleanup_wp05_test_rows(pg_engine)