cayce
73ccdf7bdf
ci / lint-test-leaks (pull_request) Canceled after 0s
scanner (read-only): - sources from CAYCE_OPS_SCANNER_SOURCES JSON (rss + page types), stdlib only - keyword scoring -> risk_score; kind via heuristics (affiliate/partnership/content_gap) - idempotent: sha1 dedupe_key; repeated scans skip existing - every proposal audited (actor daemon:scanner); source errors recorded, never fatal executor (human gate): - approve/reject audited with operator identity; only proposed -> approved - AUTONOMOUS_MODE=true adds exactly one path: policy auto-approval (kinds allowlist + daily cap, actor auto:policy) — DECISIONS D5 - plugin registry by kind; unknown kinds fall back to log_action (D9) - retries with max_attempts, then failed; every transition audited - http_post plugin inert unless CAYCE_OPS_ENABLE_HTTP_ACTIONS=true tests: 13 new (dedupe, keyword filter, autonomy bounds, retry-then-fail, blocked http_post, fallback plugin) — 20 total passing
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""executor tests — human gate, autonomy bounds, retries, audit chain."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from cayceops.components.executor import executor, plugins
|
|
from cayceops.models import AuditEntry, Proposal, ProposalStatus
|
|
|
|
|
|
def _proposal(kind="content_gap", title="Test item", **kw):
|
|
return Proposal(kind=kind, title=title, **kw)
|
|
|
|
|
|
def test_approve_requires_proposed_state(cfg, sess):
|
|
p = _proposal()
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
|
|
# second approve must fail (state is now approved)
|
|
assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
|
|
assert p.approved_by == "gustavo"
|
|
assert p.status == ProposalStatus.APPROVED
|
|
|
|
|
|
def test_reject_records_actor(cfg, sess):
|
|
p = _proposal()
|
|
sess.add(p)
|
|
sess.flush()
|
|
assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
|
|
assert p.status == ProposalStatus.REJECTED
|
|
|
|
|
|
def test_run_executes_only_approved(cfg, sess):
|
|
approved = _proposal(title="Approved item")
|
|
proposed = _proposal(title="Proposed item")
|
|
sess.add_all([approved, proposed])
|
|
sess.flush()
|
|
executor.approve(approved.id, "gustavo", cfg(), sess)
|
|
|
|
config = cfg() # AUTONOMOUS_MODE=false
|
|
assert executor.run(config, sess) == 0
|
|
|
|
assert approved.status == ProposalStatus.DONE
|
|
assert proposed.status == ProposalStatus.PROPOSED # untouched
|
|
|
|
acts = [a.action for a in sess.query(AuditEntry).order_by(AuditEntry.id).all()]
|
|
assert "execute_start" in acts and "execute_done" in acts
|
|
assert acts.count("approve") == 1
|
|
|
|
|
|
def test_autonomy_off_by_default(cfg, sess):
|
|
p = _proposal(kind="content_gap", title="Never auto-runs")
|
|
sess.add(p)
|
|
sess.flush()
|
|
config = cfg() # autonomous_mode False even with kinds configured
|
|
executor._auto_approve(config, sess)
|
|
assert p.status == ProposalStatus.PROPOSED
|
|
|
|
|
|
def test_autonomy_respects_kinds_and_daily_cap(cfg, sess):
|
|
p1 = _proposal(kind="content_gap", title="auto ok")
|
|
p2 = _proposal(kind="content_gap", title="auto over cap")
|
|
p3 = _proposal(kind="affiliate", title="kind not allowed")
|
|
sess.add_all([p1, p2, p3])
|
|
sess.flush()
|
|
|
|
config = cfg(
|
|
CAYCE_OPS_AUTONOMOUS_MODE="true",
|
|
CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
|
|
CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
|
|
)
|
|
approved = executor._auto_approve(config, sess)
|
|
assert approved == 1
|
|
assert p1.status == ProposalStatus.APPROVED
|
|
assert p2.status == ProposalStatus.PROPOSED # cap reached
|
|
assert p3.status == ProposalStatus.PROPOSED # kind not in policy
|
|
|
|
actor = sess.query(AuditEntry).filter_by(action="approve").one().actor
|
|
assert actor == "auto:policy"
|
|
|
|
|
|
def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
|
|
p = _proposal(kind="affiliate", title="Will fail")
|
|
sess.add(p)
|
|
sess.flush()
|
|
executor.approve(p.id, "gustavo", cfg(), sess)
|
|
|
|
calls = {"n": 0}
|
|
|
|
def boom(proposal, cfg, sess):
|
|
calls["n"] += 1
|
|
raise RuntimeError("upstream exploded")
|
|
|
|
monkeypatch.setitem(plugins.REGISTRY, "affiliate", boom)
|
|
config = cfg(CAYCE_OPS_MAX_ATTEMPTS="2")
|
|
|
|
executor.run(config, sess) # attempt 1 -> back to approved for retry
|
|
assert p.status == ProposalStatus.APPROVED
|
|
assert p.attempts == 1
|
|
|
|
executor.run(config, sess) # attempt 2 -> failed for good
|
|
assert p.status == ProposalStatus.FAILED
|
|
assert p.attempts == 2
|
|
assert calls["n"] == 2
|
|
|
|
errors = sess.query(AuditEntry).filter_by(result="error").all()
|
|
assert any(a.action == "fail" for a in errors)
|
|
|
|
|
|
def test_http_post_blocked_when_disabled(cfg, sess):
|
|
p = _proposal(kind="affiliate", title=" webhook target")
|
|
p.detail = {"webhook_url": "https://hooks.example.org/ingest"}
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
config = cfg() # CAYCE_OPS_ENABLE_HTTP_ACTIONS=false
|
|
with pytest.raises(PermissionError):
|
|
plugins.http_post(p, config, sess)
|
|
|
|
|
|
def test_unknown_kind_falls_back_to_log_action(cfg, sess):
|
|
p = _proposal(kind="brand_new_kind", title="No plugin yet")
|
|
sess.add(p)
|
|
sess.flush()
|
|
executor.approve(p.id, "gustavo", cfg(), sess)
|
|
|
|
assert executor.run(cfg(), sess) == 0
|
|
assert p.status == ProposalStatus.DONE
|
|
|
|
done = sess.query(AuditEntry).filter_by(action="execute_done").one()
|
|
assert done.detail["outcome"]["plugin_fallback"] is True
|