target-revenue/tests/test_onboarding_hosted.py

"""End-to-end test: scripts/trf_onboard.py against a real, running hosted
Trust Service instance (WP-0006-T07), not just FastAPI's in-process
TestClient — this is the one test in the suite that exercises an actual
HTTP round trip over a real socket, since the onboarding CLI uses
`urllib` against a real URL rather than an ASGI transport.

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

from __future__ import annotations

import json
import shutil
import subprocess
import threading
import time
import uuid
from pathlib import Path

import pytest

psycopg = pytest.importorskip("psycopg")
pytest.importorskip("fastapi")
uvicorn = pytest.importorskip("uvicorn")

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",
]

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-onboard-{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()
            token = "test-token-onboard"
            conn.execute(
                "INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
                (token, "onboard-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 {"app_dsn": app_dsn, "token": token}
    finally:
        subprocess.run(["docker", "stop", name], capture_output=True)


@pytest.fixture()
def live_server(pg_container, monkeypatch):
    monkeypatch.setenv("TRF_DATABASE_URL", pg_container["app_dsn"])
    monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "22" * 32)
    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
    if hasattr(app_module.app.state, "signing_key"):
        del app_module.app.state.signing_key

    config = uvicorn.Config(app_module.app, host="127.0.0.1", port=0, log_level="warning")
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    for _ in range(100):
        if getattr(server, "started", False):
            break
        time.sleep(0.05)
    else:
        raise RuntimeError("uvicorn server did not start in time")

    port = server.servers[0].sockets[0].getsockname()[1]
    try:
        yield f"http://127.0.0.1:{port}"
    finally:
        server.should_exit = True
        thread.join(timeout=5)


def test_onboard_cli_registers_phase_and_appends_entry_against_live_server(
    live_server, pg_container, tmp_path, monkeypatch, capsys
):
    import importlib.util

    spec = importlib.util.spec_from_file_location("trf_onboard", REPO_ROOT / "scripts" / "trf_onboard.py")
    trf_onboard = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(trf_onboard)

    manifest = golden_manifest()
    manifest["phase"]["id"] = manifest["phase"]["id"] + "-onboard-" + uuid.uuid4().hex[:6]
    manifest_path = tmp_path / "manifest.json"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")

    monkeypatch.setenv("TRF_ONBOARD_TEST_TOKEN", pg_container["token"])

    trf_onboard.main(
        [
            "register-phase",
            "--url", live_server,
            "--token-env", "TRF_ONBOARD_TEST_TOKEN",
            "--manifest", str(manifest_path),
        ]
    )

    entry = {
        "id": "trsl:entry:onboardcli0001",
        "phase": manifest["phase"]["id"],
        "type": "development-credit",
        "amount": 1000,
        "currency": "USD",
        "recognized_at": "2026-08-01T00:00:00Z",
        "evidence_reference": "confidential:evidence:onboardcli0001",
        "extension": {"id": "trsl:extension:development-license", "version": "1.0"},
    }
    entry_path = tmp_path / "entry.json"
    entry_path.write_text(json.dumps(entry), encoding="utf-8")

    trf_onboard.main(
        [
            "append-entry",
            "--url", live_server,
            "--token-env", "TRF_ONBOARD_TEST_TOKEN",
            "--phase-id", manifest["phase"]["id"],
            "--entry", str(entry_path),
        ]
    )

    capsys.readouterr()  # discard register-phase/append-entry output
    trf_onboard.main(["status", "--url", live_server, "--phase-id", manifest["phase"]["id"]])
    status = json.loads(capsys.readouterr().out)
    assert status["facts"]["cumulative_development_credit"] == 1000