target-revenue/tests/test_control_plane_app.py

"""HTTP-level integration tests for the Control Plane interactive UI
(WP-0009-T04), via FastAPI's TestClient.

Same ephemeral, disposable Postgres-via-Docker pattern as
`test_control_plane.py` (never the shared state-hub instance).

Scope disclosure: these tests exercise the FastAPI app's routing, form
handling, session auth, and rights gating at the HTTP layer — they do not
render or interact with the pages in a real browser, since no
browser-automation tool is available in this environment. The
whynot-design web components (`<wn-*>` custom elements) are not
themselves exercised; only the server-rendered HTML/session/redirect
behavior around them is.
"""

from __future__ import annotations

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

import pytest

psycopg = pytest.importorskip("psycopg")
pytest.importorskip("jinja2")
pytest.importorskip("itsdangerous")

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

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-cpapp-{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 admin_conn:
            for migration in MIGRATIONS:
                admin_conn.execute(migration.read_text(encoding="utf-8"))
                admin_conn.commit()
            admin_conn.execute(
                "INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by) "
                "VALUES (%s, %s, %s, %s, %s)",
                ("founding-admin-token", "binky", "founder", "admin", "bootstrap"),
            )
            admin_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, "admin_token": "founding-admin-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"])
    monkeypatch.setenv("TRF_CONTROL_PLANE_SECRET_KEY", "test-secret-key")
    monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "11" * 32)

    import importlib

    from target_revenue.service import control_plane_app as mod

    importlib.reload(mod)
    if hasattr(mod.app.state, "pool"):
        mod.app.state.pool.close()
        del mod.app.state.pool
    if hasattr(mod.app.state, "signing_key"):
        del mod.app.state.signing_key

    from fastapi.testclient import TestClient

    with TestClient(mod.app) as test_client:
        yield test_client

    if hasattr(mod.app.state, "pool"):
        mod.app.state.pool.close()


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


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

    admin = registry.authenticate(conn, pg_container["admin_token"])
    suffix = uuid.uuid4().hex[:8]
    tiers = {}
    for label, rights in [
        ("viewer-user", "viewer"),
        ("contributor-user", "contributor"),
        ("operator-user", "operator"),
    ]:
        cred = registry.issue_sub_credential(
            conn, licensor_id="binky", credential_label=f"{label}-{suffix}", rights=rights,
            issued_by="founder",
        )
        tiers[rights] = cred
    conn.commit()
    tiers["admin"] = admin
    return tiers


def _login(client, token):
    resp = client.post("/login", data={"token": token}, follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/"


def test_root_requires_login_redirects(client):
    resp = client.get("/", follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/login"


def test_invalid_token_rejected(client):
    resp = client.post("/login", data={"token": "not-a-real-token"}, follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/login"


def test_valid_login_then_dashboard(client, credentials):
    _login(client, credentials["viewer"].token)
    resp = client.get("/")
    assert resp.status_code == 200
    assert "Phases for binky" in resp.text


def test_viewer_cannot_reach_phase_new(client, credentials):
    _login(client, credentials["viewer"].token)
    resp = client.get("/phases/new", follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/"


def test_operator_registers_phase_and_appends_entry(client, credentials):
    _login(client, credentials["operator"].token)
    phase_id = "trsl:phase:cpapp-test-" + uuid.uuid4().hex[:8]

    resp = client.post(
        "/phases/new",
        data={
            "phase_id": phase_id,
            "milestone_release_name": "CP UI smoke test release",
            "source_revision": "abc123",
            "initial_target_amount": "1000",
            "currency": "USD",
            "future_license": "MIT",
            "degeneration_policy": "trsl:policy:linear-longstop-v0@1.0",
            "longstop_at": "2027-01-01T00:00:00Z",
            "ledger": f"examples/{phase_id}/ledger.json",
        },
        follow_redirects=False,
    )
    assert resp.status_code == 303
    assert resp.headers["location"] == f"/phases/{phase_id}"

    detail = client.get(f"/phases/{phase_id}")
    assert detail.status_code == 200
    assert phase_id in detail.text

    ledger_resp = client.post(
        f"/phases/{phase_id}/ledger",
        data={
            "entry_id": "trsl:entry:cpappledger0001",
            "amount": "500",
            "currency": "USD",
            "recognized_at": "2026-08-01T00:00:00Z",
            "evidence_reference": "confidential:evidence:cpappledger0001",
            "extension_id": "trsl:extension:development-license",
            "extension_version": "1.0",
        },
        follow_redirects=False,
    )
    assert ledger_resp.status_code == 303

    detail_after = client.get(f"/phases/{phase_id}")
    assert "trsl:entry:cpappledger0001" in detail_after.text


def test_contributor_proposes_operator_approves(client, credentials):
    _login(client, credentials["operator"].token)
    phase_id = "trsl:phase:cpapp-propose-" + uuid.uuid4().hex[:8]
    client.post(
        "/phases/new",
        data={
            "phase_id": phase_id,
            "milestone_release_name": "CP UI propose-flow release",
            "source_revision": "def456",
            "initial_target_amount": "1000",
            "currency": "USD",
            "future_license": "MIT",
            "degeneration_policy": "trsl:policy:linear-longstop-v0@1.0",
            "longstop_at": "2027-01-01T00:00:00Z",
            "ledger": f"examples/{phase_id}/ledger.json",
        },
    )
    client.post("/logout")

    _login(client, credentials["contributor"].token)
    propose_resp = client.post(
        f"/phases/{phase_id}/ledger",
        data={
            "entry_id": "trsl:entry:cpappprop0001",
            "amount": "250",
            "currency": "USD",
            "recognized_at": "2026-08-01T00:00:00Z",
            "evidence_reference": "confidential:evidence:cpappprop0001",
            "extension_id": "trsl:extension:development-license",
            "extension_version": "1.0",
        },
        follow_redirects=False,
    )
    assert propose_resp.status_code == 303
    client.post("/logout")

    _login(client, credentials["operator"].token)
    proposals_page = client.get("/proposals")
    assert proposals_page.status_code == 200
    assert phase_id in proposals_page.text


def test_admin_issues_and_revokes_credential(client, credentials):
    _login(client, credentials["admin"].token)
    issue_resp = client.post(
        "/admin/credentials",
        data={"credential_label": f"dana-{uuid.uuid4().hex[:6]}", "rights": "viewer"},
    )
    assert issue_resp.status_code == 200
    assert "New credential issued" in issue_resp.text


def test_non_admin_cannot_reach_admin_credentials(client, credentials):
    _login(client, credentials["operator"].token)
    resp = client.get("/admin/credentials", follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/"


def test_audit_log_visible_to_signed_in_user(client, credentials):
    _login(client, credentials["viewer"].token)
    resp = client.get("/audit")
    assert resp.status_code == 200


def test_form_bridge_script_present(client):
    """whynot-design's wn-input/wn-select/wn-button are not
    form-associated custom elements — their real <input>/<select>/
    <button> live inside shadow DOM, invisible to an ancestor <form>. A
    plain click on wn-button[type=submit] silently does nothing, and
    even a submitted form would carry none of the field values. This
    only regressed once (base.html's bridging script), so pin its
    presence — TestClient can't click a real button/shadow DOM, this is
    the closest offline check available without a browser-automation
    tool."""
    resp = client.get("/login")
    assert "wn-button[type=submit]" in resp.text
    assert "requestSubmit" in resp.text
    assert "data-wn-mirror-for" in resp.text