DataOps-platform/tests/integration/wp06_postgres_identities.py

"""Ephemeral, real PostgreSQL identities for WP06 migration/runtime tests.

The administrator is deliberately limited to test setup and teardown.  The
actual Alembic process runs as a fresh LOGIN, NOSUPERUSER, NOCREATEROLE
migrator.  The runtime login has only the runtime group and must
use the SECURITY DEFINER gateways.
"""

from __future__ import annotations

import os
import subprocess
import uuid
from dataclasses import dataclass
from pathlib import Path

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

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


@dataclass(frozen=True)
class Wp06DatabaseIdentities:
    migration_url: str
    runtime_url: str
    migrator: str
    runtime: str


def _alembic(url: str, command: str, revision: str) -> None:
    environment = dict(os.environ, MIGRATION_DATABASE_URL=url)
    result = subprocess.run(
        [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), command, revision],
        cwd=ROOT, env=environment, capture_output=True, text=True,
    )
    if result.returncode:
        raise AssertionError(result.stderr)


def _quoted(connection, value: str) -> str:
    return connection.dialect.identifier_preparer.quote(value)


@pytest.fixture(scope="session")
def wp06_postgres_identities() -> Wp06DatabaseIdentities:
    """Create and remove isolated low-privilege WP06 test logins.

    ``TEST_DATABASE_URL`` is a DBA-only bootstrap input.  It is never passed
    to Alembic or to a runtime repository during this fixture's lifetime.
    """
    administrator_url = os.environ.get("TEST_DATABASE_URL")
    if not administrator_url:
        pytest.skip("TEST_DATABASE_URL is required")
    suffix = uuid.uuid4().hex[:12]
    migrator = f"wp06_migrator_{suffix}"
    runtime = f"wp06_runtime_{suffix}"
    password = f"wp06_{uuid.uuid4().hex}"
    admin_engine = create_engine(administrator_url, pool_pre_ping=True)
    original_migration = os.environ.get("TEST_MIGRATION_DATABASE_URL")
    original_runtime = os.environ.get("TEST_RUNTIME_DATABASE_URL")
    try:
        with admin_engine.begin() as connection:
            quoted_migrator = _quoted(connection, migrator)
            quoted_runtime = _quoted(connection, runtime)
            connection.execute(text(
                f"CREATE ROLE {quoted_migrator} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
            ), {"password": password})
            connection.execute(text(
                f"CREATE ROLE {quoted_runtime} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
            ), {"password": password})
            # DBA setup only: stable roles, memberships, and the ownership
            # prerequisites that a real deployment role-init performs.
            connection.execute(text(
                f"GRANT dataops_edge_evidence_owner, dataops_trusted_delivery_owner, dataops_trusted_delivery_writer, dataops_agent_runtime_owner, dataops_tenant_foundation_owner, dataops_tenant_control TO {quoted_migrator}"
            ))
            connection.execute(text(f"GRANT dataops_app_runtime, dataops_agent_runtime TO {quoted_runtime}"))
            connection.execute(text(
                "GRANT USAGE, CREATE ON SCHEMA public TO dataops_trusted_delivery_owner, dataops_trusted_delivery_writer, dataops_agent_runtime_owner, dataops_tenant_foundation_owner"
            ))
            connection.execute(text("GRANT USAGE ON SCHEMA public TO dataops_agent_runtime"))
            connection.execute(text(
                f"GRANT REFERENCES ON TABLE public.users, public.governed_agents TO {quoted_migrator}"
            ))
            # The no-superuser migrator needs only FK-definition privileges on
            # pre-existing P3 tables.  It still receives no data DML/DDL owner
            # escape hatch outside the dedicated owner/writer memberships.
            connection.execute(text(
                f"GRANT REFERENCES ON TABLE public.data_incidents, public.data_observability_alerts, public.governance_tasks, public.agent_credentials TO {quoted_migrator}"
            ))
            connection.execute(text(f"""
                GRANT SELECT, UPDATE ON TABLE public.governance_tasks TO {quoted_migrator} WITH GRANT OPTION;
                GRANT SELECT ON TABLE public.governance_task_reviews,
                  public.agent_tool_grants, public.governed_agents TO {quoted_migrator} WITH GRANT OPTION
                ; GRANT SELECT, UPDATE ON TABLE public.agent_credentials TO {quoted_migrator} WITH GRANT OPTION
            """))
            roles = connection.execute(text("""
                SELECT rolname, rolsuper, rolcreaterole
                  FROM pg_roles WHERE rolname IN (:migrator, :runtime)
            """), {"migrator": migrator, "runtime": runtime}).mappings().all()
            assert {(row["rolname"], row["rolsuper"], row["rolcreaterole"]) for row in roles} == {
                (migrator, False, False), (runtime, False, False)
            }

        parsed = make_url(administrator_url)
        migration_url = parsed.set(username=migrator, password=password).render_as_string(hide_password=False)
        runtime_url = parsed.set(username=runtime, password=password).render_as_string(hide_password=False)
        os.environ["TEST_MIGRATION_DATABASE_URL"] = migration_url
        os.environ["TEST_RUNTIME_DATABASE_URL"] = runtime_url
        yield Wp06DatabaseIdentities(migration_url, runtime_url, migrator, runtime)
    finally:
        if original_migration is None:
            os.environ.pop("TEST_MIGRATION_DATABASE_URL", None)
        else:
            os.environ["TEST_MIGRATION_DATABASE_URL"] = original_migration
        if original_runtime is None:
            os.environ.pop("TEST_RUNTIME_DATABASE_URL", None)
        else:
            os.environ["TEST_RUNTIME_DATABASE_URL"] = original_runtime
        with admin_engine.begin() as connection:
            for role in (migrator, runtime):
                quoted_role = _quoted(connection, role)
                connection.execute(text(f"REASSIGN OWNED BY {quoted_role} TO dataops"))
                connection.execute(text(f"DROP OWNED BY {quoted_role}"))
                connection.execute(text(f"DROP ROLE IF EXISTS {quoted_role}"))
        admin_engine.dispose()


@pytest.fixture(scope="module")
def wp06_head(wp06_postgres_identities) -> Wp06DatabaseIdentities:
    """Start each WP06 PostgreSQL module at 480, then exercise 480 -> head.

    The DBA-only reset is isolated test teardown/setup.  The actual forward
    chain is always run by the low-privilege migrator and the fixture leaves
    the database at head for later targeted tests.
    """
    administrator_url = os.environ["TEST_DATABASE_URL"]
    _alembic(administrator_url, "downgrade", "20260811_480")
    _alembic(wp06_postgres_identities.migration_url, "upgrade", "head")
    yield wp06_postgres_identities