target-revenue/tests/test_licensor_credentials.py

"""Integration tests for WP-0009-T02 (per-human sub-credentials).

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

from __future__ import annotations

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

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-creds-{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()
            # A first, "founding" credential for the tenant, as if issued
            # at Licensor-onboarding time (analogous to earlier test
            # modules' plain single-token setup).
            conn.execute(
                "INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by) "
                "VALUES (%s, %s, %s, %s, %s)",
                ("founding-token", "binky", "founding-admin", "admin", "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


def test_issue_sub_credential_resolves_same_licensor_id(conn, pg_container):
    from target_revenue import registry

    founding = registry.authenticate(conn, pg_container["founding_token"])
    assert founding.licensor_id == "binky"

    alice = registry.issue_sub_credential(
        conn, licensor_id="binky", credential_label="alice", rights="operator", issued_by="founding-admin"
    )
    conn.commit()
    assert alice.licensor_id == "binky"
    assert alice.credential_label == "alice"
    assert alice.rights == "operator"
    assert alice.token != pg_container["founding_token"]

    reauth = registry.authenticate(conn, alice.token)
    assert reauth.licensor_id == "binky"
    assert reauth.credential_label == "alice"
    assert reauth.rights == "operator"


def test_duplicate_active_label_for_same_licensor_rejected(conn):
    from target_revenue import registry

    registry.issue_sub_credential(
        conn, licensor_id="binky", credential_label="bob", rights="viewer", issued_by="founding-admin"
    )
    conn.commit()

    with pytest.raises(registry.RegistrationError, match="already exists"):
        registry.issue_sub_credential(
            conn, licensor_id="binky", credential_label="bob", rights="operator", issued_by="founding-admin"
        )


def test_revoked_credential_cannot_authenticate(conn):
    from target_revenue import registry

    carol = registry.issue_sub_credential(
        conn, licensor_id="binky", credential_label="carol", rights="contributor", issued_by="founding-admin"
    )
    conn.commit()

    registry.revoke_sub_credential(conn, carol.token)
    conn.commit()

    with pytest.raises(registry.RegistrationError, match="unrecognized or revoked"):
        registry.authenticate(conn, carol.token)


def test_revoking_unknown_token_is_not_an_error(conn):
    from target_revenue import registry

    registry.revoke_sub_credential(conn, "not-a-real-token")  # must not raise
    conn.commit()


def test_invalid_rights_tier_rejected(conn):
    from target_revenue import registry

    with pytest.raises(registry.RegistrationError, match="rights must be one of"):
        registry.issue_sub_credential(
            conn, licensor_id="binky", credential_label="dave", rights="superuser", issued_by="founding-admin"
        )


def test_has_right_ordinal_comparison():
    from target_revenue import registry

    assert registry.has_right("admin", "viewer")
    assert registry.has_right("operator", "contributor")
    assert not registry.has_right("viewer", "operator")
    assert registry.has_right("contributor", "contributor")


def test_ledger_entry_attribution_recorded(conn, pg_container):
    from target_revenue import ledger, registry
    from target_revenue.service import keys

    alice = registry.issue_sub_credential(
        conn, licensor_id="binky", credential_label="alice-ledger", rights="operator", issued_by="founding-admin"
    )
    conn.commit()

    manifest = golden_manifest()
    manifest["phase"]["id"] = manifest["phase"]["id"] + "-attribution-" + uuid.uuid4().hex[:6]
    registry.register_phase_manifest(conn, alice, manifest)
    conn.commit()

    signing_key = keys.load_signing_key()
    entry = {
        "id": "trsl:entry:attrtest0001",
        "phase": manifest["phase"]["id"],
        "type": "development-credit",
        "amount": 1,
        "currency": manifest["phase"]["initial_target"]["currency"],
        "recognized_at": "2026-08-01T00:00:00Z",
        "evidence_reference": "confidential:evidence:attrtest0001",
        "extension": {"id": "trsl:extension:development-license", "version": "1.0"},
    }
    ledger.append_entry(conn, alice, manifest["phase"]["id"], entry, signing_key)
    conn.commit()

    attribution = ledger.get_ledger_attribution(conn, manifest["phase"]["id"])
    assert attribution == [
        {"entry_id": "trsl:entry:attrtest0001", "credential_label": "alice-ledger", "rights": "operator"}
    ]

    # The exported ledger entry itself carries no attribution field —
    # unchanged, schema-conformant, exactly as portable as before this task.
    exported = ledger.get_ledger(conn, manifest["phase"]["id"])
    assert "credential_label" not in exported[0]
    assert "submitted_by" not in exported[0]


def test_application_role_cannot_update_licensors_directly(pg_container):
    with psycopg.connect(pg_container["app_dsn"]) as app_conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            app_conn.execute(
                "UPDATE licensors SET rights = 'admin' WHERE token = 'founding-token'"
            )
        app_conn.rollback()