cayce
a2de419467
- README: the stack belongs to Cayce, an autonomous entity operating under the Próspera ZEDE agentic-companies jurisdiction pilot; the experiment is managed by Cayce itself - tests: approver actor renamed to the generic label operator - zero occurrences of personal names in the tree (verified with grep)
145 lines
4.0 KiB
Python
145 lines
4.0 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 = "operator"
|
|
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 == "operator"
|
|
|
|
|
|
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="operator", 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 == "operator"
|
|
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"
|
|
|
|
|
|
def test_audit_chain_and_verification(sess):
|
|
from cayceops.models import verify_audit_chain
|
|
|
|
p = Proposal(kind="other", title="chain")
|
|
sess.add(p)
|
|
sess.flush()
|
|
for i in range(3):
|
|
audit(sess, actor=f"actor-{i}", action="step", proposal_id=p.id, detail={"i": i})
|
|
sess.flush()
|
|
|
|
ok, broken = verify_audit_chain(sess)
|
|
assert ok and broken is None
|
|
|
|
# tamper with one entry -> chain breaks at that id
|
|
first = sess.query(AuditEntry).order_by(AuditEntry.id).first()
|
|
first.actor = "tampered"
|
|
ok, broken = verify_audit_chain(sess)
|
|
assert not ok and broken == first.id
|
|
|
|
|
|
def test_concurrent_writers_keep_chain_verifiable(db_engine):
|
|
"""Round-2 point 4: serialized writes -> no forked chains."""
|
|
import threading
|
|
|
|
import cayceops.db as appdb
|
|
|
|
errors: list[Exception] = []
|
|
|
|
def writer(worker: int, n: int):
|
|
try:
|
|
for i in range(n):
|
|
with appdb.session_scope() as s:
|
|
audit(s, actor=f"w{worker}", action=f"step{i}", detail={"worker": worker})
|
|
except Exception as exc: # noqa: BLE001
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=writer, args=(w, 25)) for w in range(4)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors, errors
|
|
s = appdb.session_factory()()
|
|
total = s.query(AuditEntry).count()
|
|
assert total == 100
|
|
from cayceops.models import verify_audit_chain
|
|
|
|
ok, broken = verify_audit_chain(s)
|
|
assert ok and broken is None
|