cayce-ops/tests/conftest.py

"""Shared pytest fixtures.

Tests run against a throwaway SQLite database (fast, zero external services).
The production path is Postgres via Alembic; ``make migrate-check`` smoke-tests
the migration chain on SQLite in CI.
All fixture data is invented — never real merchant or account data.
"""

from __future__ import annotations

import os
from pathlib import Path
from types import SimpleNamespace

import pytest
from coincurve import PublicKey

# test operator keypair (invented; allowlisted in .gitleaks.toml)
OPERATOR_SECRET = bytes.fromhex("22" * 32)
OPERATOR_PUBKEY = PublicKey.from_secret(OPERATOR_SECRET).format().hex()[2:]

# Config must see the test DB before cayceops.config/db get imported.
os.environ.setdefault("CAYCE_OPS_DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("CAYCE_OPS_AUTONOMOUS_MODE", "false")


@pytest.fixture()
def db_engine(tmp_path: Path):
    """A fresh schema per test, on a file-backed SQLite DB (shared in-process)."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from cayceops.db import reset_engine
    from cayceops.models import Base

    url = f"sqlite:///{tmp_path}/test.db"
    eng = create_engine(url, future=True, connect_args={"check_same_thread": False})
    Base.metadata.create_all(eng)
    factory = sessionmaker(bind=eng, expire_on_commit=False, future=True)

    # point the app at the same engine/session factory
    reset_engine()
    import cayceops.db as appdb

    appdb._engine = eng
    appdb._factory = factory
    yield eng
    eng.dispose()
    reset_engine()


@pytest.fixture()
def sess(db_engine):
    """A session bound to the fresh schema (rolls back at test end)."""
    import cayceops.db as appdb

    s = appdb.session_factory()()
    yield s
    s.rollback()
    s.close()


@pytest.fixture()
def signer():
    """Signs approval messages with the test operator key (22*32, deployment
    'itest'). Mirrors exactly what the operator CLI does."""
    from coincurve import PrivateKey

    from cayceops.models import MailDirection
    from cayceops.signing import (
        approval_message,
        content_content_hash,
        mail_content_hash,
        proposal_content_hash,
    )

    secret = OPERATOR_SECRET

    def _sig(msg: bytes) -> str:
        return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()

    def proposal(p, deployment="itest"):
        chash = proposal_content_hash(p.kind, p.title, p.url, p.detail)
        return _sig(approval_message("proposal", deployment, p.id, chash))

    def mail(record, deployment="itest"):
        to = (
            record.from_addr
            if record.direction == MailDirection.INBOUND
            else record.to_addr
        )
        chash = mail_content_hash(to, record.subject, record.draft_reply)
        return _sig(approval_message("mail", deployment, record.id, chash))

    def content(item, deployment="itest"):
        chash = content_content_hash(item.channel, item.caption, item.link)
        return _sig(approval_message("content", deployment, item.id, chash))

    def raw(sig_bytes: bytes) -> str:
        return _sig(sig_bytes)

    def msg(item_type: str, deployment: str, item_id: int, chash: str) -> bytes:
        from cayceops.signing import approval_message

        return approval_message(item_type, deployment, item_id, chash)

    def content_hash(item, deployment="itest") -> str:
        from cayceops.signing import content_content_hash

        return content_content_hash(item.channel, item.caption, item.link)

    return SimpleNamespace(
        proposal=proposal, mail=mail, content=content, raw=raw, msg=msg,
        content_hash=content_hash,
    )


@pytest.fixture()
def cfg(monkeypatch):
    """A Config snapshot with test defaults (no network, no autonomy)."""
    from cayceops.config import Config

    defaults = {
        "CAYCE_OPS_DATABASE_URL": "sqlite:///:memory:",
        "CAYCE_OPS_LLM_PROVIDER": "null",
        "CAYCE_OPS_ENABLE_FACEBOOK": "false",
        "CAYCE_OPS_ENABLE_INSTAGRAM": "false",
        "CAYCE_OPS_NOSTR_RELAYS": "",
        "CAYCE_OPS_MEDIA_DIR": "media",
        # signing defaults (audit round 2): tests operate in a signed
        # deployment unless they explicitly clear these
        "CAYCE_OPS_DEPLOYMENT_ID": "itest",
        "CAYCE_OPS_OPERATOR_PUBKEY": OPERATOR_PUBKEY,
    }

    def _apply(**overrides):
        merged = {**defaults, **{k: str(v) for k, v in overrides.items()}}
        for k, v in merged.items():
            monkeypatch.setenv(k, v)
        return Config()

    return _apply