target-revenue/tests/test_ledger_hosting.py

"""Integration tests for WP-0006-T04 (hosted Target Ledger append API).

Same ephemeral, disposable Postgres-via-Docker pattern as
test_registry_hosting.py (never the shared state-hub instance). The single
property this file tests most aggressively, per the task's own framing:
appending entries through the hosted API and then folding the exported
result offline (`fold.fold_outstanding_target`) must produce exactly the
same Outstanding Target a hosted metrics/attestation component would later
report — a hosted service that silently diverges from the offline fold is
the highest-risk defect this component could introduce.
"""

from __future__ import annotations

import shutil
import subprocess
import time
import uuid
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",
]

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-ledger-{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) VALUES (%s, %s)",
                (token, "acme-corp"),
            )
            other_token = "test-token-other"
            conn.execute(
                "INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
                (other_token, "other-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,
            "other_token": other_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}"}


@pytest.fixture()
def registered_phase(client, pg_container):
    manifest = golden_manifest()
    manifest["phase"]["id"] = manifest["phase"]["id"] + "-ledger-test-" + uuid.uuid4().hex[:6]
    resp = client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"]))
    assert resp.status_code == 201, resp.text
    return manifest


def _entry(phase_id: str, entry_id: str, entry_type: str, amount: float, recognized_at: str):
    return {
        "id": entry_id,
        "phase": phase_id,
        "type": entry_type,
        "amount": amount,
        "currency": "USD",
        "recognized_at": recognized_at,
        "evidence_reference": f"confidential:evidence:{entry_id}",
        "extension": {"id": "trsl:extension:development-license", "version": "1.0"},
    }


def test_append_and_hash_chain_links(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    e1 = _entry(phase_id, "trsl:entry:ledgertest0001", "development-credit", 1000, "2026-08-01T00:00:00Z")
    r1 = client.post(f"/phases/{phase_id}/ledger", json=e1, headers=auth_headers(pg_container["token"]))
    assert r1.status_code == 201, r1.text
    stored1 = r1.json()
    assert stored1["previous_entry_hash"] == "GENESIS"
    assert "signature" in stored1

    e2 = _entry(phase_id, "trsl:entry:ledgertest0002", "development-credit", 500, "2026-08-02T00:00:00Z")
    r2 = client.post(f"/phases/{phase_id}/ledger", json=e2, headers=auth_headers(pg_container["token"]))
    assert r2.status_code == 201, r2.text
    stored2 = r2.json()

    from target_revenue import hashing

    assert stored2["previous_entry_hash"] == hashing.entry_hash(stored1)
    # Full chain, as exported, must independently verify.
    hashing.verify_chain([stored1, stored2])


def test_client_cannot_supply_previous_entry_hash_or_signature(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    forged = _entry(phase_id, "trsl:entry:ledgertest0003", "development-credit", 1, "2026-08-01T00:00:00Z")
    forged["previous_entry_hash"] = "GENESIS"
    resp = client.post(f"/phases/{phase_id}/ledger", json=forged, headers=auth_headers(pg_container["token"]))
    assert resp.status_code == 422
    assert "must not be supplied" in resp.json()["detail"]


def test_other_licensor_cannot_append_to_this_phase(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    e = _entry(phase_id, "trsl:entry:ledgertest0004", "development-credit", 1, "2026-08-01T00:00:00Z")
    resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["other_token"]))
    assert resp.status_code == 422
    assert "not authorized" in resp.json()["detail"]


def test_currency_mismatch_rejected(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    e = _entry(phase_id, "trsl:entry:ledgertest0005", "development-credit", 1, "2026-08-01T00:00:00Z")
    e["currency"] = "EUR"
    resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
    assert resp.status_code == 422
    assert "currency" in resp.json()["detail"]


def test_hosted_ledger_export_folds_identically_to_offline_computation(
    client, pg_container, registered_phase
):
    """The property T04's task description calls out as highest-risk:
    hosted append + export, folded offline, must match what was appended."""
    from target_revenue import fold, hashing

    phase_id = registered_phase["phase"]["id"]
    plan = [
        ("development-credit", 25000),
        ("development-credit", 10000),
        ("remission-credit", 5000),
        ("credit-reversal", 2000),
    ]
    for i, (entry_type, amount) in enumerate(plan, start=1):
        e = _entry(
            phase_id,
            f"trsl:entry:ledgertestfold{i:04d}",
            entry_type,
            amount,
            f"2026-08-0{i}T00:00:00Z",
        )
        if entry_type == "credit-reversal":
            e["reverses"] = "trsl:entry:ledgertestfold0001"
        resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
        assert resp.status_code == 201, resp.text

    export = client.get(f"/phases/{phase_id}/ledger")
    assert export.status_code == 200
    entries = export.json()
    assert len(entries) == len(plan)

    # Chain integrity survives the round trip through Postgres/JSONB/API.
    hashing.verify_chain(entries)

    result = fold.fold_outstanding_target(
        registered_phase["phase"]["initial_target"]["amount"], entries
    )
    expected_development_credit = 25000 + 10000 - 2000
    expected_remission_credit = 5000
    assert result.development_credit == expected_development_credit
    assert result.remission_credit == expected_remission_credit
    assert result.outstanding_target == max(
        0.0,
        registered_phase["phase"]["initial_target"]["amount"]
        - expected_development_credit
        - expected_remission_credit,
    )


def test_public_key_endpoint_unauthenticated_and_verifies_signature(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    e = _entry(phase_id, "trsl:entry:ledgertestpk0001", "development-credit", 1, "2026-08-01T00:00:00Z")
    stored = client.post(
        f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])
    ).json()

    pk_resp = client.get("/public-key")
    assert pk_resp.status_code == 200
    assert pk_resp.json()["algorithm"] == "ed25519"

    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

    from target_revenue import hashing

    public_key = Ed25519PublicKey.from_public_bytes(
        bytes.fromhex(pk_resp.json()["public_key_hex"])
    )
    assert hashing.verify_record_signature(stored, stored["signature"], public_key)


def test_application_role_cannot_update_or_delete_ledger_entries(pg_container):
    with psycopg.connect(pg_container["app_dsn"]) as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("UPDATE ledger_entries SET signature = 'x' WHERE entry_id = 'nonexistent'")
        conn.rollback()
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("DELETE FROM ledger_entries WHERE entry_id = 'nonexistent'")
        conn.rollback()


def test_duplicate_entry_id_rejected(client, pg_container, registered_phase):
    phase_id = registered_phase["phase"]["id"]
    e = _entry(phase_id, "trsl:entry:ledgertestdup0001", "development-credit", 1, "2026-08-01T00:00:00Z")
    first = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
    assert first.status_code == 201
    second = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
    assert second.status_code == 422
    assert "already exists" in second.json()["detail"]


def test_metrics_endpoint_unauthenticated_and_matches_offline_computation(
    client, pg_container, registered_phase
):
    """WP-0006-T05: the hosted /metrics response must match what
    target_revenue.metrics.compute_metrics computes offline from the same
    exported Manifest + Ledger — the metrics-layer analogue of T04's
    hosted/offline fold-agreement test."""
    from target_revenue import metrics as metrics_module

    phase_id = registered_phase["phase"]["id"]
    e = _entry(phase_id, "trsl:entry:ledgertestmetrics0001", "development-credit", 4000, "2026-08-01T00:00:00Z")
    resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
    assert resp.status_code == 201, resp.text

    # No Authorization header at all: metrics are public per FR-9/FR-10.
    metrics_resp = client.get(f"/phases/{phase_id}/metrics")
    assert metrics_resp.status_code == 200
    hosted = metrics_resp.json()

    entries = client.get(f"/phases/{phase_id}/ledger").json()
    as_of = metrics_module.datetime.fromisoformat(hosted["as_of"])
    offline = metrics_module.compute_metrics(registered_phase, entries, as_of)

    assert hosted == offline
    assert hosted["facts"]["cumulative_development_credit"] == 4000