cayce
34a5f331ac
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)
93 lines
2.4 KiB
Python
93 lines
2.4 KiB
Python
"""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"
|