skeleton: package, schema, config, CLI, alembic, CI, deploy, docs
ci / lint-test-leaks (pull_request) Canceled after 0s
ci / lint-test-leaks (pull_request) Canceled after 0s
- cayceops core: env-only config (CAYCE_OPS_*), engine/session, ORM models with state machines (proposal/mail/content) and audited writes - provider-agnostic LLM interface (null provider default, stdlib openai-compat) - Typer CLI with lazy component loading (scanner/executor/maild/publisher/monitor) - alembic 0001 initial schema; sqlite-portable JSON columns - tests: 7 passing on throwaway sqlite; fixtures invented, never real data - ci: ruff + alembic smoke + pytest + gitleaks(full history) in .gitea/workflows + .github/workflows; local gate = make check - deploy: docker-compose (postgres 16) + hardened systemd units/timers - docs: DECISIONS D1-D9; README project (grant text retired)
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
"""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
|
||||
|
||||
import pytest
|
||||
|
||||
# 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 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",
|
||||
}
|
||||
|
||||
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
|
||||
@@ -0,0 +1,23 @@
|
||||
"""CLI smoke tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from cayceops import __version__
|
||||
from cayceops.cli import app
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def test_version():
|
||||
result = runner.invoke(app, ["version"])
|
||||
assert result.exit_code == 0
|
||||
assert __version__ in result.output
|
||||
|
||||
|
||||
def test_help_lists_components():
|
||||
result = runner.invoke(app, ["--help"])
|
||||
assert result.exit_code == 0
|
||||
for sub in ("scanner", "executor", "maild", "publisher", "monitor", "db"):
|
||||
assert sub in result.output
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Schema-level tests: models persist, enums hold, audit helper writes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import select
|
||||
|
||||
from cayceops.models import (
|
||||
AuditEntry,
|
||||
ContentItem,
|
||||
DaemonRun,
|
||||
MailMessage,
|
||||
Proposal,
|
||||
ProposalSource,
|
||||
ProposalStatus,
|
||||
audit,
|
||||
)
|
||||
|
||||
|
||||
def test_proposal_defaults(sess):
|
||||
p = Proposal(kind="affiliate", title="Test merchant program")
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
|
||||
assert p.id is not None
|
||||
assert p.status == ProposalStatus.PROPOSED
|
||||
assert p.source == ProposalSource.SCANNER
|
||||
assert p.attempts == 0
|
||||
assert p.created_at.tzinfo is not None
|
||||
|
||||
|
||||
def test_proposal_status_transitions(sess):
|
||||
p = Proposal(kind="content_gap", title="Guide gap: Rio")
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
|
||||
p.status = ProposalStatus.APPROVED
|
||||
p.approved_by = "gustavo"
|
||||
p.approved_at = datetime.now(UTC)
|
||||
p.status = ProposalStatus.RUNNING
|
||||
p.status = ProposalStatus.DONE
|
||||
sess.flush()
|
||||
|
||||
loaded = sess.scalar(select(Proposal).where(Proposal.id == p.id))
|
||||
assert loaded.status == ProposalStatus.DONE
|
||||
assert loaded.approved_by == "gustavo"
|
||||
|
||||
|
||||
def test_dedupe_key_unique(sess):
|
||||
sess.add(Proposal(kind="affiliate", title="A", dedupe_key="k-1"))
|
||||
sess.flush()
|
||||
sess.add(Proposal(kind="affiliate", title="B", dedupe_key="k-1"))
|
||||
try:
|
||||
sess.flush()
|
||||
raised = False
|
||||
except sa.exc.IntegrityError:
|
||||
raised = True
|
||||
sess.rollback()
|
||||
assert raised, "duplicate dedupe_key must be rejected"
|
||||
|
||||
|
||||
def test_audit_helper_writes(sess):
|
||||
p = Proposal(kind="other", title="x")
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
audit(sess, actor="gustavo", action="approve", proposal_id=p.id, detail={"note": "ok"})
|
||||
sess.flush()
|
||||
|
||||
rows = list(sess.scalars(select(AuditEntry)).all())
|
||||
assert len(rows) == 1
|
||||
assert rows[0].actor == "gustavo"
|
||||
assert rows[0].proposal_id == p.id
|
||||
|
||||
|
||||
def test_mail_and_content_and_runs(sess):
|
||||
m = MailMessage(
|
||||
direction="in",
|
||||
subject="Welcome to the partner program",
|
||||
classification="approval",
|
||||
confidence=0.9,
|
||||
draft_status="drafted",
|
||||
)
|
||||
c = ContentItem(theme="lightning-abroad", channel="nostr", caption="gm")
|
||||
r = DaemonRun(daemon="scanner", status="ok", detail={"found": 3})
|
||||
sess.add_all([m, c, r])
|
||||
sess.flush()
|
||||
|
||||
assert m.id and c.id and r.id
|
||||
assert m.draft_status == "drafted"
|
||||
assert c.status == "draft"
|
||||
Reference in New Issue
Block a user