cayce-ops/tests/integration/test_pg_roles.py

"""Integration tests against Postgres 16: roles.sql + one full daemon cycle
per role (audit round 2, point 3).

Skipped unless a Postgres is reachable at TEST_PG_DSN (default: the
docker-compose service). Roles are applied by piping deploy/roles.sql
through psql WITH the mandatory -v password variables — exercising the same
path an operator uses; the test skips if the psql client is not installed.

Run:  uv run pytest -m integration -v
"""

from __future__ import annotations

import os
import subprocess
from pathlib import Path

import pytest

pytestmark = pytest.mark.integration

PG_DSN = os.environ.get(
    "TEST_PG_DSN",
    "postgresql://cayce:ci-only-not-a-secret@127.0.0.1:5432/cayce_ops",
)
ROLES_SQL = Path(__file__).resolve().parents[2] / "deploy" / "roles.sql"

ROLE_PASSWORDS = {
    "scanner_password": "itest-scanner-pw",
    "executor_password": "itest-executor-pw",
    "maild_password": "itest-maild-pw",
    "publisher_password": "itest-publisher-pw",
    "monitor_password": "itest-monitor-pw",
    "approver_password": "itest-approver-pw",
}


def _pg_available() -> bool:
    try:
        import psycopg

        with psycopg.connect(PG_DSN, connect_timeout=3):
            return True
    except Exception:  # noqa: BLE001
        return False


def _psql_available() -> bool:
    try:
        return (
            subprocess.run(
                ["psql", "--version"], capture_output=True, timeout=10
            ).returncode
            == 0
        )
    except (OSError, subprocess.SubprocessError):
        return False


@pytest.fixture(scope="module")
def roles_applied():
    if not _pg_available():
        pytest.skip("Postgres not reachable (set TEST_PG_DSN)")
    if not _psql_available():
        pytest.skip("psql client not installed")

    # fresh slate on an admin connection
    import psycopg

    admin = psycopg.connect(PG_DSN, autocommit=True)
    with admin.cursor() as cur:
        cur.execute(
            """
            DROP TABLE IF EXISTS audit_log, mail_messages, content_items,
              daemon_runs, proposals, alembic_version CASCADE;
            DROP TYPE IF EXISTS proposalsource, proposalstatus, maildirection,
              mailclassification, draftstatus, contentstatus CASCADE;
            DROP FUNCTION IF EXISTS cayce_ops_approval_gate() CASCADE;
            """
        )
    admin.close()

    # apply alembic chain (imports app; config must point at PG first)
    os.environ["CAYCE_OPS_DATABASE_URL"] = PG_DSN.replace(
        "postgresql://", "postgresql+psycopg://"
    )
    import subprocess as sp

    env = {**os.environ, "CAYCE_OPS_DATABASE_URL": os.environ["CAYCE_OPS_DATABASE_URL"]}
    # upgrade -> downgrade -> upgrade round-trip on Postgres (audit round 2,
    # point 6): exercises 0002.downgrade and 0003.downgrade for real
    sp.run(["uv", "run", "alembic", "upgrade", "head"], check=True, env=env)
    sp.run(["uv", "run", "alembic", "downgrade", "0001"], check=True, env=env)
    sp.run(["uv", "run", "alembic", "upgrade", "head"], check=True, env=env)

    # apply roles.sql the operator way: psql -v <password vars>
    cmd = ["psql", PG_DSN, "-v", "ON_ERROR_STOP=1", "-f", str(ROLES_SQL)]
    for var, value in ROLE_PASSWORDS.items():
        cmd += ["-v", f"{var}={value}"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    assert result.returncode == 0, f"roles.sql failed:\n{result.stderr}"

    # re-apply WITHOUT variables must fail (audit requirement: no silent run)
    bad = subprocess.run(
        ["psql", PG_DSN, "-v", "ON_ERROR_STOP=1", "-f", str(ROLES_SQL)],
        capture_output=True,
        text=True,
        timeout=60,
    )
    assert bad.returncode != 0, "roles.sql must refuse to run without vars"

    yield


def _connect_as(role: str):
    """Connect with the role's credentials on the same DB as PG_DSN."""
    import psycopg

    key = f"{role.split('_', 1)[1]}_password"
    after_scheme = PG_DSN.split("://", 1)[1]
    userinfo, hostport_db = after_scheme.split("@", 1)
    hostport, _db = hostport_db.split("/", 1)
    dsn = f"postgresql://{role}:{ROLE_PASSWORDS[key]}@{hostport}/{_db}"
    return psycopg.connect(dsn, autocommit=True)


def _must_fail(conn, sql: str, because: str):
    import psycopg

    try:
        with conn.cursor() as cur:
            cur.execute(sql)
    except psycopg.Error as exc:
        assert because.lower() in str(exc).lower(), f"{because} — got: {exc}"
        conn.rollback()
        return
    pytest.fail(f"expected failure: {because}")


def test_roles_full_daemon_cycle(roles_applied):

    # --- scanner role: proposes (INSERT proposals) ---
    with _connect_as("cayce_scanner") as conn, conn.cursor() as cur:
        cur.execute(
            """INSERT INTO proposals (source, kind, title, url, detail,
                 match_score, status, attempts, created_at, updated_at)
               VALUES ('SCANNER', 'partnership', 'Acme partner', NULL, '{}',
                 0.9, 'PROPOSED', 0, now(), now()) RETURNING id"""
        )
        proposal_id = cur.fetchone()[0]
        cur.execute(
            "INSERT INTO audit_log (ts, actor, action, detail, result) "
            "VALUES (now(), 'daemon:scanner', 'propose', '{}', 'ok')"
        )
        cur.execute(
            "INSERT INTO daemon_runs (daemon, started_at, status, detail) "
            "VALUES ('scanner', now(), 'ok', '{}')"
        )

    # scanner cannot approve: no UPDATE grant at all, and the trigger would
    # still block status=APPROVED for non-approver roles
    scanner = _connect_as("cayce_scanner")
    _must_fail(
        scanner,
        f"UPDATE proposals SET status='APPROVED' WHERE id={proposal_id}",
        because="permission denied",
    )
    scanner.close()

    # executor cannot approve either (trigger), and cannot write signatures
    executor = _connect_as("cayce_executor")
    # the signature column is outside the executor's column grant: the whole
    # UPDATE is denied before the trigger even runs
    fake_sig = "ab" * 64  # invented 128-hex signature
    _must_fail(
        executor,
        f"UPDATE proposals SET status='APPROVED', operator_signature='{fake_sig}' "
        f"WHERE id={proposal_id}",
        because="permission denied",
    )
    # executor CAN run its cycle once the approver approved
    with executor.cursor() as cur:
        cur.execute(
            f"UPDATE proposals SET status='RUNNING', attempts=attempts+1, "
            f"updated_at=now() WHERE id={proposal_id}"
        )
        cur.execute(
            f"UPDATE proposals SET status='DONE', updated_at=now() WHERE id={proposal_id}"
        )
        cur.execute(
            "INSERT INTO daemon_runs (daemon, started_at, status, detail) "
            "VALUES ('executor', now(), 'ok', '{}')"
        )
    executor.close()

    # --- approver: the only role that can approve + sign ---
    sig = "c9" * 64  # invented 128-hex signature for the permission test
    with _connect_as("cayce_approver") as conn, conn.cursor() as cur:
        cur.execute(
            f"UPDATE proposals SET status='APPROVED', approved_by='operator', "
            f"approved_at=now(), operator_signature='{sig}' WHERE id={proposal_id}"
        )

    # --- maild role: drafts + sends, cannot approve ---
    with _connect_as("cayce_maild") as conn, conn.cursor() as cur:
        cur.execute(
            """INSERT INTO mail_messages (direction, message_id, from_addr,
                 subject, body, draft_reply, draft_status, confidence)
               VALUES ('OUTBOUND', '<itest@x>', 'desk@acme.example.org', 'Hello',
                 'Body', 'Body', 'DRAFTED', 0) RETURNING id"""
        )
        mail_id = cur.fetchone()[0]
    maild = _connect_as("cayce_maild")
    _must_fail(
        maild,
        f"UPDATE mail_messages SET draft_status='APPROVED' WHERE id={mail_id}",
        because="only cayce_approver may set mail draft_status=approved",
    )
    _must_fail(
        maild,
        f"UPDATE mail_messages SET operator_signature='{sig}' WHERE id={mail_id}",
        because="permission denied",  # signature column outside maild's grant
    )
    # the approver approves; then maild can mark it sent
    with _connect_as("cayce_approver") as conn, conn.cursor() as cur:
        cur.execute(
            "UPDATE mail_messages SET draft_status='APPROVED', "
            "approved_by='operator', approved_hash='"
            + "a" * 64
            + f"', operator_signature='{sig}' WHERE id={mail_id}"
        )
    with _connect_as("cayce_maild") as conn, conn.cursor() as cur:
        cur.execute(
            f"UPDATE mail_messages SET draft_status='SENT' WHERE id={mail_id}"
        )
    maild.close()

    # --- publisher role: plans content, cannot approve ---
    with _connect_as("cayce_publisher") as conn, conn.cursor() as cur:
        cur.execute(
            """INSERT INTO content_items (theme, channel, caption, status,
                 attempts, created_at)
               VALUES ('atm-fees', 'nostr', 'caption', 'DRAFT', 0, now())
               RETURNING id"""
        )
        item_id = cur.fetchone()[0]
    publisher = _connect_as("cayce_publisher")
    _must_fail(
        publisher,
        f"UPDATE content_items SET status='APPROVED' WHERE id={item_id}",
        because="only cayce_approver may set content status=approved",
    )
    # approver approves; publisher publishes
    with _connect_as("cayce_approver") as conn, conn.cursor() as cur:
        cur.execute(
            "UPDATE content_items SET status='APPROVED', approved_hash='"
            + "b" * 64
            + f"', operator_signature='{sig}' WHERE id={item_id}"
        )
    with _connect_as("cayce_publisher") as conn, conn.cursor() as cur:
        cur.execute(
            f"UPDATE content_items SET status='PUBLISHED', published_at=now() "
            f"WHERE id={item_id}"
        )
    publisher.close()

    # --- monitor role: read-only + audit anchors ---
    monitor = _connect_as("cayce_monitor")
    with monitor.cursor() as cur:
        cur.execute("SELECT count(*) FROM proposals")
        cur.fetchone()
        cur.execute(
            "INSERT INTO audit_log (ts, actor, action, detail, result) "
            "VALUES (now(), 'daemon:monitor', 'anchor', '{}', 'ok')"
        )
    # monitor CAN book its runs, but cannot touch proposals
    with monitor.cursor() as cur:
        cur.execute(
            "INSERT INTO daemon_runs (daemon, started_at, status, detail) "
            "VALUES ('monitor', now(), 'ok', '{}')"
        )
    _must_fail(
        monitor,
        "INSERT INTO proposals (source, kind, title, detail, match_score, status, "
        "attempts, created_at, updated_at) "
        "VALUES ('SCANNER', 'other', 'x', '{}', 0, 'PROPOSED', 0, now(), now())",
        because="permission denied",
    )
    monitor.close()