target-revenue/tests/test_registry_hosting.py

"""Integration tests for WP-0006-T03 (hosted Phase/Extension Registry).

Spins up an ephemeral, disposable PostgreSQL container via `docker run`
(never the shared `infra-postgres-1`/`custodian` instance used by the state
hub), applies `migrations/0001_registries.sql`, and exercises the hosted
API through FastAPI's TestClient. Skipped automatically if Docker is not
available, so the offline WP-0002 suite (`tests/test_*.py`) is unaffected.

Requires the `service-dev` extra: `pip install -e ".[service-dev]"`.
"""

from __future__ import annotations

import shutil
import subprocess
import time
import uuid

import pytest

psycopg = pytest.importorskip("psycopg")
pytest.importorskip("fastapi")
from fastapi.testclient import TestClient  # noqa: E402

from conftest import golden_extension, golden_manifest  # noqa: E402

REPO_ROOT_MIGRATIONS = (
    __import__("pathlib").Path(__file__).resolve().parents[1] / "migrations" / "0001_registries.sql"
)

pytestmark = pytest.mark.skipif(
    shutil.which("docker") is None, reason="docker not available"
)


@pytest.fixture(scope="module")
def pg_container():
    name = f"trf-test-pg-{uuid.uuid4().hex[:8]}"
    subprocess.run(
        [
            "docker", "run", "--rm", "-d",
            "--name", name,
            "-e", "POSTGRES_PASSWORD=postgres",
            "-e", "POSTGRES_DB=target_revenue_test",
            "-p", "127.0.0.1::5432",
            "postgres:16-alpine",
        ],
        check=True, capture_output=True,
    )
    try:
        port_out = subprocess.run(
            ["docker", "port", name, "5432/tcp"], check=True, capture_output=True, text=True
        ).stdout.strip()
        host_port = port_out.split(":")[-1]
        dsn = f"host=127.0.0.1 port={host_port} dbname=target_revenue_test user=postgres password=postgres"

        for _ in range(60):
            try:
                with psycopg.connect(dsn, connect_timeout=1):
                    break
            except psycopg.OperationalError:
                time.sleep(0.5)
        else:
            raise RuntimeError("postgres container did not become ready in time")

        with psycopg.connect(dsn) as conn:
            conn.execute(REPO_ROOT_MIGRATIONS.read_text(encoding="utf-8"))
            conn.commit()
            token = "test-token-acme"
            conn.execute(
                "INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
                (token, "acme-corp"),
            )
            conn.commit()

        app_dsn = (
            f"host=127.0.0.1 port={host_port} dbname=target_revenue_test "
            f"user=trf_app password=changeme-in-deployment"
        )
        yield {"admin_dsn": dsn, "app_dsn": app_dsn, "token": token}
    finally:
        subprocess.run(["docker", "stop", name], capture_output=True)


@pytest.fixture()
def client(pg_container, monkeypatch):
    monkeypatch.setenv("TRF_DATABASE_URL", pg_container["app_dsn"])
    from target_revenue.service import app as app_module

    if hasattr(app_module.app.state, "pool"):
        app_module.app.state.pool.close()
        del app_module.app.state.pool
    with TestClient(app_module.app) as c:
        yield c


def auth_headers(pg_container):
    return {"Authorization": f"Bearer {pg_container['token']}"}


def test_register_and_read_golden_manifest(client, pg_container):
    manifest = golden_manifest()
    resp = client.post("/phases", json=manifest, headers=auth_headers(pg_container))
    assert resp.status_code == 201, resp.text
    assert resp.json()["status"] == "registered"

    read = client.get(f"/phases/{manifest['phase']['id']}")
    assert read.status_code == 200
    assert read.json() == manifest


def test_reject_non_conformant_manifest(client, pg_container):
    broken = golden_manifest()
    del broken["phase"]["initial_target"]
    resp = client.post("/phases", json=broken, headers=auth_headers(pg_container))
    assert resp.status_code == 422
    assert "initial_target" in resp.json()["detail"]


def test_cannot_reregister_same_phase_id(client, pg_container):
    manifest = golden_manifest()
    manifest["phase"]["id"] = manifest["phase"]["id"] + "-dup-test"
    first = client.post("/phases", json=manifest, headers=auth_headers(pg_container))
    assert first.status_code == 201
    second = client.post("/phases", json=manifest, headers=auth_headers(pg_container))
    assert second.status_code == 422
    assert "already registered" in second.json()["detail"]


def test_unknown_token_rejected(client):
    manifest = golden_manifest()
    resp = client.post(
        "/phases", json=manifest, headers={"Authorization": "Bearer not-a-real-token"}
    )
    assert resp.status_code == 401


def test_register_extension_and_promote_canonical(client, pg_container):
    ext = golden_extension("development-license")
    resp = client.post("/extensions", json=ext, headers=auth_headers(pg_container))
    assert resp.status_code == 201, resp.text
    assert resp.json()["status"] == "registered"

    read = client.get(f"/extensions/{ext['id']}/{ext['version']}")
    assert read.json()["status"] == "registered"

    import psycopg as _psycopg
    from target_revenue import registry as registry_module

    with _psycopg.connect(pg_container["admin_dsn"]) as admin_conn:
        registry_module.promote_extension_canonical(
            admin_conn, ext["id"], ext["version"], approved_by="Bernd"
        )
        admin_conn.commit()

    read_after = client.get(f"/extensions/{ext['id']}/{ext['version']}")
    assert read_after.json()["status"] == "canonical"


def test_application_role_cannot_update_or_delete_settled_manifest(pg_container):
    """Database-level enforcement check (ADR-0002 compensating guardrail 3):
    the trf_app role must have no UPDATE/DELETE grant on phase_manifests at
    all, independent of what the API layer happens to expose."""
    with psycopg.connect(pg_container["app_dsn"]) as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute(
                "UPDATE phase_manifests SET manifest = manifest WHERE phase_id = 'nonexistent'"
            )
        conn.rollback()
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("DELETE FROM phase_manifests WHERE phase_id = 'nonexistent'")
        conn.rollback()


def test_application_role_cannot_update_extensions_directly(pg_container):
    with psycopg.connect(pg_container["app_dsn"]) as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute(
                "UPDATE extensions SET status = 'canonical' WHERE extension_id = 'nonexistent'"
            )
        conn.rollback()