target-revenue/tests/test_remission_hosting.py

"""Hosted integration tests for WP-0013 remission automation.

Same ephemeral Postgres-via-Docker pattern as test_ledger_hosting.py.
"""

from __future__ import annotations

import shutil
import subprocess
import time
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

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

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-remission-{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-acme"
            conn.execute(
                "INSERT INTO licensors (token, licensor_id, credential_label, rights) "
                "VALUES (%s, %s, %s, %s)",
                (token, "acme-corp", "admin-acme", "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, "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"])
    monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "11" * 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
    with TestClient(app_module.app) as c:
        yield c


def auth_headers(token: str):
    return {"Authorization": f"Bearer {token}"}


def test_apply_remission_writes_entry_and_is_idempotent(client, pg_container):
    from target_revenue import fold

    manifest = golden_manifest()
    manifest["phase"]["id"] = "trsl:phase:remission-test-" + uuid.uuid4().hex[:8]
    # Multi-year longstop so wall-clock drift between the two apply calls
    # stays under MIN_REMISSION_AMOUNT (idempotency floor). Backdate t0 so
    # R(now) is still a material fraction of T0.
    now = datetime.now(timezone.utc)
    manifest["phase"]["longstop_at"] = (now + timedelta(days=365 * 5)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    r = client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"]))
    assert r.status_code == 201, r.text
    phase_id = manifest["phase"]["id"]

    with psycopg.connect(pg_container["admin_dsn"]) as conn:
        conn.execute(
            "UPDATE phase_manifests SET registered_at = %s WHERE phase_id = %s",
            (now - timedelta(days=365), phase_id),
        )
        conn.commit()

    first = client.post(
        f"/phases/{phase_id}/remission", headers=auth_headers(pg_container["token"])
    )
    assert first.status_code == 201, first.text
    body = first.json()
    assert body["status"] == "appended"
    entry = body["entry"]
    assert entry["type"] == "remission-credit"
    assert entry["amount"] > 0
    assert entry["extension"]["id"] == "trsl:policy:linear-longstop-v0"

    second = client.post(
        f"/phases/{phase_id}/remission", headers=auth_headers(pg_container["token"])
    )
    assert second.status_code == 201, second.text
    assert second.json()["status"] == "up_to_date"
    assert second.json()["entry"] is None

    entries = client.get(f"/phases/{phase_id}/ledger").json()
    remissions = [e for e in entries if e["type"] == "remission-credit"]
    assert len(remissions) == 1

    result = fold.fold_outstanding_target(manifest["phase"]["initial_target"]["amount"], entries)
    assert result.remission_credit == remissions[0]["amount"]

    # Attribution: submitted by system:policy-engine, not the human token.
    with psycopg.connect(pg_container["admin_dsn"]) as conn:
        row = conn.execute(
            """
            SELECT li.credential_label
              FROM ledger_entries le
              JOIN licensors li ON li.token = le.submitted_by_token
             WHERE le.phase_id = %s
            """,
            (phase_id,),
        ).fetchone()
        assert row[0] == "system:policy-engine"


def test_metrics_include_remission_forecasts(client, pg_container):
    manifest = golden_manifest()
    manifest["phase"]["id"] = "trsl:phase:remission-metrics-" + uuid.uuid4().hex[:8]
    client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
    phase_id = manifest["phase"]["id"]

    metrics = client.get(f"/phases/{phase_id}/metrics").json()
    assert metrics["facts"]["activated_at"] is not None
    assert "remission_if_applied_now" in metrics["forecasts"]
    assert "next_scheduled_remission_at" in metrics["forecasts"]