target-revenue/tests/test_forgejo_hubs.py

"""Integration tests for WP-0015-T05 (`forgejo_hubs` registry).

Same ephemeral, disposable Postgres-via-Docker pattern as the other
hosted test modules (never the shared state-hub instance).
"""

from __future__ import annotations

import copy
import shutil
import subprocess
import time
import uuid
from pathlib import Path

import pytest

psycopg = pytest.importorskip("psycopg")

from conftest import golden_manifest  # noqa: E402

REPO_ROOT = Path(__file__).resolve().parents[1]
MIGRATIONS = [
    REPO_ROOT / "migrations" / "0001_registries.sql",
    REPO_ROOT / "migrations" / "0002_ledger.sql",
    REPO_ROOT / "migrations" / "0003_attestations.sql",
    REPO_ROOT / "migrations" / "0004_breach_records.sql",
    REPO_ROOT / "migrations" / "0005_licensor_credentials.sql",
    REPO_ROOT / "migrations" / "0006_control_plane.sql",
    REPO_ROOT / "migrations" / "0007_forgejo_hubs.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-hubs-{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:
            for migration in MIGRATIONS:
                conn.execute(migration.read_text(encoding="utf-8"))
                conn.commit()
            conn.execute(
                "INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by) "
                "VALUES (%s, %s, %s, %s, %s)",
                ("founding-token", "binky", "founder", "operator", "bootstrap"),
            )
            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, "founding_token": "founding-token"}
    finally:
        subprocess.run(["docker", "stop", name], capture_output=True)


@pytest.fixture()
def conn(pg_container):
    with psycopg.connect(pg_container["app_dsn"]) as connection:
        yield connection


@pytest.fixture()
def licensor(conn, pg_container):
    from target_revenue import registry

    return registry.authenticate(conn, pg_container["founding_token"])


def _manifest_with_hub(hub_slug: str, hub_uri: str, phase_suffix: str) -> dict:
    manifest = copy.deepcopy(golden_manifest())
    manifest["phase"]["id"] = manifest["phase"]["id"] + "-hubtest-" + phase_suffix
    manifest["phase"]["milestone_release"]["repo_hub"] = hub_slug
    manifest["phase"]["milestone_release"]["repo_hub_uri"] = hub_uri
    return manifest


def test_registering_phase_auto_creates_hub_row(conn, licensor):
    from target_revenue import registry

    suffix = uuid.uuid4().hex[:8]
    manifest = _manifest_with_hub(f"hub-{suffix}", "https://forgejo.example.invalid", suffix)
    registry.register_phase_manifest(conn, licensor, manifest)
    conn.commit()

    hub = registry.get_forgejo_hub(conn, f"hub-{suffix}")
    assert hub is not None
    assert hub["service_uri"] == "https://forgejo.example.invalid"


def test_unknown_hub_returns_none(conn):
    from target_revenue import registry

    assert registry.get_forgejo_hub(conn, "no-such-hub") is None


def test_second_phase_same_hub_does_not_overwrite(conn, licensor):
    """ON CONFLICT DO NOTHING: a hub already seen is left alone by the
    auto-populate trigger -- only correct_forgejo_hub_uri() may change it."""
    from target_revenue import registry

    suffix = uuid.uuid4().hex[:8]
    hub_slug = f"hub-{suffix}"
    manifest_a = _manifest_with_hub(hub_slug, "https://first-seen.example.invalid", suffix + "a")
    registry.register_phase_manifest(conn, licensor, manifest_a)
    conn.commit()

    manifest_b = _manifest_with_hub(hub_slug, "https://different-uri.example.invalid", suffix + "b")
    registry.register_phase_manifest(conn, licensor, manifest_b)
    conn.commit()

    hub = registry.get_forgejo_hub(conn, hub_slug)
    assert hub["service_uri"] == "https://first-seen.example.invalid"


def test_correct_forgejo_hub_uri_updates_existing_hub(conn, licensor):
    from target_revenue import registry

    suffix = uuid.uuid4().hex[:8]
    hub_slug = f"hub-{suffix}"
    manifest = _manifest_with_hub(hub_slug, "https://old-domain.example.invalid", suffix)
    registry.register_phase_manifest(conn, licensor, manifest)
    conn.commit()

    registry.correct_forgejo_hub_uri(conn, hub_slug, "https://new-domain.example.invalid")
    conn.commit()

    hub = registry.get_forgejo_hub(conn, hub_slug)
    assert hub["service_uri"] == "https://new-domain.example.invalid"


def test_correct_forgejo_hub_uri_unknown_hub_is_rejected(conn):
    from target_revenue import registry

    with pytest.raises(registry.RegistrationError, match="unknown forgejo hub"):
        registry.correct_forgejo_hub_uri(conn, "never-registered", "https://x.example.invalid")


def test_application_role_cannot_update_forgejo_hubs_directly(pg_container):
    with psycopg.connect(pg_container["app_dsn"]) as app_conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            app_conn.execute(
                "UPDATE forgejo_hubs SET service_uri = 'https://tampered.invalid' "
                "WHERE hub_slug = 'anything'"
            )
        app_conn.rollback()