From 73ccdf7bdf19c5732bbae9c2a39770bb22f2192c Mon Sep 17 00:00:00 2001 From: Cayce Date: Mon, 19 Oct 2026 21:42:34 +0000 Subject: [PATCH] scanner + executor: discovery files proposals; execution gated on approval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cayceops/components/executor/__init__.py | 1 + cayceops/components/executor/executor.py | 246 +++++++++++++++++++++++ cayceops/components/executor/plugins.py | 78 +++++++ cayceops/components/scanner/__init__.py | 1 + cayceops/components/scanner/scanner.py | 209 +++++++++++++++++++ tests/test_executor.py | 133 ++++++++++++ tests/test_scanner.py | 112 +++++++++++ 7 files changed, 780 insertions(+) create mode 100644 cayceops/components/executor/__init__.py create mode 100644 cayceops/components/executor/executor.py create mode 100644 cayceops/components/executor/plugins.py create mode 100644 cayceops/components/scanner/__init__.py create mode 100644 cayceops/components/scanner/scanner.py create mode 100644 tests/test_executor.py create mode 100644 tests/test_scanner.py diff --git a/cayceops/components/executor/__init__.py b/cayceops/components/executor/__init__.py new file mode 100644 index 0000000..e5eab83 --- /dev/null +++ b/cayceops/components/executor/__init__.py @@ -0,0 +1 @@ +"""executor package.""" diff --git a/cayceops/components/executor/executor.py b/cayceops/components/executor/executor.py new file mode 100644 index 0000000..0851b3e --- /dev/null +++ b/cayceops/components/executor/executor.py @@ -0,0 +1,246 @@ +"""executor — executes APPROVED proposals through registered action plugins. + +Guarantees +---------- +- Only ``approved`` proposals execute (``AUTONOMOUS_MODE=false`` default). +- With ``AUTONOMOUS_MODE=true`` the executor gains exactly one capability: + bounded auto-approval of ``proposed`` items whose ``kind`` is in + ``CAYCE_OPS_AUTONOMOUS_KINDS``, capped by ``CAYCE_OPS_AUTONOMOUS_DAILY_CAP``, + recorded with actor ``auto:policy`` (DECISIONS D5). +- Every transition writes an ``audit_log`` row: who approved, what ran, + plugin output, success/failure. +- Unknown proposal kinds fall back to the ``log_action`` plugin (never a + network side effect). See DECISIONS D9. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import func, select + +from cayceops.components.executor import plugins as plugin_registry +from cayceops.config import Config +from cayceops.db import session_scope +from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, audit + +Actor = str + + +# --- human transitions ------------------------------------------------------- + + +def approve(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int: + """Human approves a proposal. Audited with the operator's identity.""" + return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess) + + +def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int: + """Human rejects a proposal. Audited with the operator's identity.""" + return _decide(proposal_id, actor, ProposalStatus.REJECTED, cfg, sess) + + +def _decide( + proposal_id: int, + actor: Actor, + decision: ProposalStatus, + cfg: Config | None, + sess, +) -> int: + owned = False + if sess is None: + outer = session_scope() + sess = outer.__enter__() + owned = True + try: + proposal = sess.get(Proposal, proposal_id) + if proposal is None: + return 1 + if proposal.status != ProposalStatus.PROPOSED: + verb = "approve" if decision == ProposalStatus.APPROVED else "reject" + audit( + sess, + actor=actor, + action=verb, + proposal_id=proposal_id, + result="error", + detail={"reason": f"invalid state: {proposal.status.value}"}, + ) + return 1 + proposal.status = decision + if decision == ProposalStatus.APPROVED: + proposal.approved_by = actor + proposal.approved_at = datetime.now(UTC) + verb = "approve" if decision == ProposalStatus.APPROVED else "reject" + audit(sess, actor=actor, action=verb, proposal_id=proposal_id) + return 0 + finally: + if owned: + outer.__exit__(None, None, None) + + +# --- autonomous policy (the ONLY non-human approval path, D5) ----------------- + + +def _auto_approve(cfg: Config, sess) -> int: + """Auto-approve proposed items per policy. Returns count approved.""" + if not cfg.autonomous_mode or not cfg.autonomous_kinds: + return 0 + midnight = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) + used = ( + sess.query(func.count(AuditEntry.id)) + .filter( + AuditEntry.actor == "auto:policy", + AuditEntry.action == "approve", + AuditEntry.ts >= midnight, + ) + .scalar() + ) + budget = max(0, cfg.autonomous_daily_cap - int(used or 0)) + approved = 0 + if budget == 0: + return 0 + candidates = ( + sess.query(Proposal) + .filter( + Proposal.status == ProposalStatus.PROPOSED, + Proposal.kind.in_(cfg.autonomous_kinds), + ) + .order_by(Proposal.risk_score.asc(), Proposal.id.asc()) + .limit(budget) + .all() + ) + for proposal in candidates: + proposal.status = ProposalStatus.APPROVED + proposal.approved_by = "auto:policy" + proposal.approved_at = datetime.now(UTC) + audit( + sess, + actor="auto:policy", + action="approve", + proposal_id=proposal.id, + detail={ + "reason": "autonomous_mode policy", + "kind": proposal.kind, + "risk_score": proposal.risk_score, + "daily_cap": cfg.autonomous_daily_cap, + }, + ) + approved += 1 + return approved + + +# --- execution ---------------------------------------------------------------- + + +def execute_one(proposal: Proposal, cfg: Config, sess) -> bool: + """Run the plugin for one approved proposal. Returns True on success.""" + proposal.status = ProposalStatus.RUNNING + proposal.attempts += 1 + audit( + sess, + actor="daemon:executor", + action="execute_start", + proposal_id=proposal.id, + detail={"plugin": proposal.kind, "attempt": proposal.attempts}, + ) + plugin = plugin_registry.REGISTRY.get(proposal.kind) + try: + if plugin is None: + outcome: dict[str, Any] = plugin_registry.log_action(proposal, cfg, sess) + outcome["plugin_fallback"] = True + outcome["note"] = f"no plugin registered for kind {proposal.kind!r}" + else: + outcome = plugin(proposal, cfg, sess) + except Exception as exc: # noqa: BLE001 — plugin boundary: capture for audit + audit( + sess, + actor="daemon:executor", + action="execute_error", + proposal_id=proposal.id, + result="error", + detail={"error": str(exc)[:300], "attempt": proposal.attempts}, + ) + if proposal.attempts >= cfg.max_attempts: + proposal.status = ProposalStatus.FAILED + audit( + sess, + actor="daemon:executor", + action="fail", + proposal_id=proposal.id, + result="error", + detail={"attempts": proposal.attempts}, + ) + else: + proposal.status = ProposalStatus.APPROVED # retry next run + return False + + proposal.status = ProposalStatus.DONE + audit( + sess, + actor="daemon:executor", + action="execute_done", + proposal_id=proposal.id, + detail={"outcome": outcome}, + ) + return True + + +def run(cfg: Config | None = None, sess=None) -> int: + """CLI entry: auto-approve per policy, then execute all approved items.""" + from cayceops.config import load + from cayceops.db import engine as _engine + + cfg = cfg or load() + owned = False + if sess is None: + outer = session_scope() + sess = outer.__enter__() + owned = True + try: + run_row = DaemonRun(daemon="executor") + sess.add(run_row) + sess.flush() + try: + auto = _auto_approve(cfg, sess) + pending = ( + sess.query(Proposal) + .filter(Proposal.status == ProposalStatus.APPROVED) + .order_by(Proposal.approved_at.asc()) + .all() + ) + done = failed = 0 + for proposal in pending: + if execute_one(proposal, cfg, sess): + done += 1 + else: + failed += 1 + run_row.status = "ok" + run_row.detail = {"auto_approved": auto, "done": done, "failed": failed} + except Exception as exc: # noqa: BLE001 — daemon boundary + run_row.status = "error" + run_row.detail = {"error": str(exc)[:300]} + run_row.finished_at = datetime.now(UTC) + return 1 + run_row.finished_at = datetime.now(UTC) + return 0 + finally: + if owned: + outer.__exit__(None, None, None) + _engine().dispose() + + +def select_approved(sess) -> list[Proposal]: + """Helper for tests/tools: currently approved proposals, oldest first.""" + return list( + sess.scalars( + select(Proposal) + .where(Proposal.status == ProposalStatus.APPROVED) + .order_by(Proposal.approved_at.asc()) + ).all() + ) + + +ActionPlugin = Callable[[Proposal, Config, Any], dict] diff --git a/cayceops/components/executor/plugins.py b/cayceops/components/executor/plugins.py new file mode 100644 index 0000000..f560185 --- /dev/null +++ b/cayceops/components/executor/plugins.py @@ -0,0 +1,78 @@ +"""executor action plugins. + +A plugin maps a proposal ``kind`` to a side effect. Plugins are repo code — +they are reviewed like everything else and must: + +- take ``(proposal, cfg, sess)`` and return a JSON-safe dict (the audit outcome) +- do their own permission checks (see ``http_post`` + ENABLE_HTTP_ACTIONS) +- raise on failure (the executor captures the exception into the audit log) +""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from typing import Any + +from cayceops.config import Config +from cayceops.models import Proposal, audit + +REGISTRY: dict[str, Any] = {} + + +def register(kind: str): + """Decorator: register an action plugin for a proposal kind.""" + + def deco(fn): + REGISTRY[kind] = fn + return fn + + return deco + + +@register("log_action") +def log_action(proposal: Proposal, cfg: Config, sess) -> dict: + """No-side-effect plugin: record the execution in the audit log only.""" + return {"action": "logged", "title": proposal.title[:200]} + + +@register("http_post") +def http_post(proposal: Proposal, cfg: Config, sess) -> dict: + """POST the proposal payload to a webhook. + + Inert unless ``CAYCE_OPS_ENABLE_HTTP_ACTIONS=true``. Target URL comes from + the proposal detail (``detail['webhook_url']``) — operator-controlled data. + """ + if not cfg.enable_http_actions: + audit( + sess, + actor="daemon:executor", + action="http_post_blocked", + proposal_id=proposal.id, + result="error", + detail={"reason": "CAYCE_OPS_ENABLE_HTTP_ACTIONS is false"}, + ) + raise PermissionError("http_post disabled (CAYCE_OPS_ENABLE_HTTP_ACTIONS=false)") + + url = (proposal.detail or {}).get("webhook_url", "") + if not url: + raise ValueError("proposal detail.webhook_url missing") + payload = json.dumps( + {"proposal_id": proposal.id, "kind": proposal.kind, "title": proposal.title} + ).encode() + req = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=20.0) as resp: # noqa: S310 + return {"action": "http_post", "status": resp.status} + + +# proposals from the scanner default to content_gap/affiliate/partnership kinds; +# until dedicated plugins exist for those kinds they must not do anything +# network-visible. register more plugins here as the operator approves them. +for _safe_kind in ("affiliate", "partnership", "content_gap", "other"): + REGISTRY.setdefault(_safe_kind, log_action) diff --git a/cayceops/components/scanner/__init__.py b/cayceops/components/scanner/__init__.py new file mode 100644 index 0000000..3f6e9dd --- /dev/null +++ b/cayceops/components/scanner/__init__.py @@ -0,0 +1 @@ +"""scanner package.""" diff --git a/cayceops/components/scanner/scanner.py b/cayceops/components/scanner/scanner.py new file mode 100644 index 0000000..4b62580 --- /dev/null +++ b/cayceops/components/scanner/scanner.py @@ -0,0 +1,209 @@ +"""scanner — opportunity discovery. Proposes; never executes. + +Sources come from ``CAYCE_OPS_SCANNER_SOURCES`` (JSON array): + [{"type": "rss", "url": "https://example.org/feed.xml"}, + {"type": "page", "url": "https://example.org/partners"}] + +Every discovered candidate becomes a ``proposals`` row with status ``proposed`` +and a stable ``dedupe_key`` (sha1 of url-or-title) so repeated scans stay +idempotent. The scanner performs no side effects beyond writing proposals and +audit entries. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import urllib.error +import urllib.request +from datetime import UTC, datetime +from html.parser import HTMLParser +from xml.etree import ElementTree + +from cayceops.config import Config +from cayceops.db import session_scope +from cayceops.models import DaemonRun, Proposal, ProposalSource, ProposalStatus, audit + +# kind classification heuristics (checked in order) +_KIND_PATTERNS = [ + ("partnership", re.compile(r"partner|collaborat|integrat", re.I)), + ("affiliate", re.compile(r"affiliate|referral|commission|revenue\s*share", re.I)), +] +DEFAULT_KIND = "content_gap" + + +class _LinkCollector(HTMLParser): + """Minimal HTML parser: page + anchor (href, text) pairs.""" + + def __init__(self) -> None: + super().__init__() + self.title: str = "" + self.links: list[tuple[str, str]] = [] + self._in_title = False + self._in_anchor = False + self._href = "" + self._text: list[str] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str]]) -> None: + if tag == "title": + self._in_title = True + elif tag == "a": + self._in_anchor = True + self._href = dict(attrs).get("href", "") + self._text = [] + + def handle_endtag(self, tag: str) -> None: + if tag == "title": + self._in_title = False + elif tag == "a" and self._in_anchor: + self._in_anchor = False + text = " ".join("".join(self._text).split()) + if self._href and text: + self.links.append((self._href, text)) + + def handle_data(self, data: str) -> None: + if self._in_title: + self.title += data + if self._in_anchor: + self._text.append(data) + + +def _fetch(url: str, cfg: Config, timeout: float = 20.0) -> bytes: + req = urllib.request.Request( + url, headers={"User-Agent": cfg.scanner_user_agent}, method="GET" + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 + return resp.read() + + +def _parse_rss(body: bytes) -> list[tuple[str, str, str]]: + """Return (title, link, description) tuples from an RSS/Atom feed.""" + out: list[tuple[str, str, str]] = [] + root = ElementTree.fromstring(body) + ns = {"atom": "http://www.w3.org/2005/Atom"} + for item in root.iter("item"): # RSS 2.0 + title = item.findtext("title") or "" + link = item.findtext("link") or "" + desc = re.sub(r"<[^>]+>", " ", item.findtext("description") or "") + out.append((" ".join(title.split()), link, " ".join(desc.split())[:500])) + for entry in root.findall("atom:entry", ns): # Atom + title = entry.findtext("atom:title", namespaces=ns) or "" + link_el = entry.find("atom:link", ns) + link = link_el.get("href", "") if link_el is not None else "" + desc = re.sub(r"<[^>]+>", " ", entry.findtext("atom:summary", namespaces=ns) or "") + out.append((" ".join(title.split()), link, " ".join(desc.split())[:500])) + return out + + +def _parse_page(body: bytes) -> list[tuple[str, str, str]]: + """Return (title=anchor text, link=href, description=page title) tuples.""" + parser = _LinkCollector() + parser.feed(body.decode("utf-8", errors="replace")) + page_title = " ".join(parser.title.split()) + return [(text, href, f"linked from page: {page_title}") for href, text in parser.links] + + +def _classify(title: str, description: str) -> str: + blob = f"{title}\n{description}" + for kind, pattern in _KIND_PATTERNS: + if pattern.search(blob): + return kind + return DEFAULT_KIND + + +def _score(title: str, description: str, keywords: list[str]) -> float: + blob = f"{title}\n{description}".lower() + hits = sum(1 for kw in keywords if kw.lower() in blob) + return round(min(hits / max(1, len(keywords)), 1.0), 3) + + +def scan(cfg: Config, sess, fetch=_fetch) -> dict: + """Run one scan pass against all configured sources. Returns run stats.""" + sources = json.loads(cfg.scanner_sources or "[]") + stats = {"sources": len(sources), "candidates": 0, "proposed": 0, "skipped_existing": 0} + existing = { + key + for (key,) in sess.query(Proposal.dedupe_key).filter( + Proposal.dedupe_key.isnot(None) + ) + } + + for src in sources: + url = src.get("url", "") + stype = src.get("type", "page") + if not url: + continue + try: + body = fetch(url, cfg) + except (urllib.error.URLError, OSError) as exc: + stats.setdefault("source_errors", []).append({"url": url, "error": str(exc)[:200]}) + continue + items = _parse_rss(body) if stype == "rss" else _parse_page(body) + stats["candidates"] += len(items) + + for title, link, desc in items: + text = title or link + if not text: + continue + keywords = cfg.scanner_keywords + score = _score(title or link, desc, keywords) + if score <= 0 and stype != "page": + continue # feeds must match at least one keyword; pages keep links + dedupe = hashlib.sha1((link or title).encode()).hexdigest()[:32] # noqa: S324 + if dedupe in existing: + stats["skipped_existing"] += 1 + continue + proposal = Proposal( + source=ProposalSource.SCANNER, + kind=_classify(text, desc), + title=text[:512], + url=(link or None)[:1024] if (link or None) else None, + detail={ + "source_type": stype, + "source_url": url, + "summary": desc, + "scanned_at": datetime.now(UTC).isoformat(), + }, + risk_score=score, + status=ProposalStatus.PROPOSED, + dedupe_key=dedupe, + ) + sess.add(proposal) + sess.flush() + audit( + sess, + actor="daemon:scanner", + action="propose", + proposal_id=proposal.id, + detail={"kind": proposal.kind, "score": score, "source": stype}, + ) + existing.add(dedupe) + stats["proposed"] += 1 + return stats + + +def run(cfg: Config | None = None) -> int: + """CLI entry: one scan pass, wrapped in a DaemonRun record.""" + from cayceops.config import load + from cayceops.db import engine + + cfg = cfg or load() + try: + with session_scope() as sess: + run_row = DaemonRun(daemon="scanner") + sess.add(run_row) + sess.flush() + try: + stats = scan(cfg, sess) + except Exception as exc: # noqa: BLE001 — daemon boundary: record and fail cleanly + run_row.status = "error" + run_row.detail = {"error": str(exc)[:300]} + run_row.finished_at = datetime.now(UTC) + return 1 + run_row.status = "ok" + run_row.detail = stats + run_row.finished_at = datetime.now(UTC) + return 0 + finally: + engine().dispose() diff --git a/tests/test_executor.py b/tests/test_executor.py new file mode 100644 index 0000000..52b9364 --- /dev/null +++ b/tests/test_executor.py @@ -0,0 +1,133 @@ +"""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 diff --git a/tests/test_scanner.py b/tests/test_scanner.py new file mode 100644 index 0000000..54e87ba --- /dev/null +++ b/tests/test_scanner.py @@ -0,0 +1,112 @@ +"""scanner tests — discovery, classification, dedupe. No network (injected fetch).""" + +from __future__ import annotations + +import json + +from cayceops.components.scanner import scanner +from cayceops.models import Proposal, ProposalStatus + +RSS = b"""<?xml version="1.0"?> +<rss><channel> + <item><title>Acme launches affiliate program for hotels + https://acme.example.org/affiliates + Earn commission on every booking. + Weekly crypto digest + https://news.example.org/digest + Everything unrelated. +""" + +PAGE = b"""Partners + Become a payment partner + API docs +""" + + +def _fake_fetch(responses): + def fetch(url, cfg, timeout=20.0): + return responses[url] + + return fetch + + +SOURCES = [ + {"type": "rss", "url": "https://feeds.example.org/news.xml"}, + {"type": "page", "url": "https://pay.example.org/partners"}, +] + + +def _cfg(cfg, **kw): + return cfg( + CAYCE_OPS_SCANNER_SOURCES=json.dumps(SOURCES), + CAYCE_OPS_SCANNER_KEYWORDS="affiliate,partner,bitcoin,commission", + **kw, + ) + + +def test_scan_files_proposals_with_kind_and_score(cfg, sess): + config = _cfg(cfg) + stats = scanner.scan(config, sess, fetch=_fake_fetch({ + "https://feeds.example.org/news.xml": RSS, + "https://pay.example.org/partners": PAGE, + })) + + assert stats["proposed"] == 3 # affiliate item + both page links (pages keep all) + proposals = sess.query(Proposal).all() + by_title = {p.title: p for p in proposals} + + affiliate = by_title["Acme launches affiliate program for hotels"] + assert affiliate.kind == "affiliate" + assert affiliate.status == ProposalStatus.PROPOSED + assert affiliate.risk_score > 0 + assert affiliate.dedupe_key + + partner = by_title["Become a payment partner"] + assert partner.kind == "partnership" + + +def test_scan_is_idempotent_via_dedupe(cfg, sess): + config = _cfg(cfg) + scanner.scan(config, sess, fetch=_fake_fetch({ + "https://feeds.example.org/news.xml": RSS, + "https://pay.example.org/partners": PAGE, + })) + stats2 = scanner.scan(config, sess, fetch=_fake_fetch({ + "https://feeds.example.org/news.xml": RSS, + "https://pay.example.org/partners": PAGE, + })) + assert stats2["proposed"] == 0 + assert stats2["skipped_existing"] == 3 + + +def test_scan_filters_keywordless_feed_items(cfg, sess): + config = _cfg(cfg) + scanner.scan(config, sess, fetch=_fake_fetch({ + "https://feeds.example.org/news.xml": RSS, + "https://pay.example.org/partners": PAGE, + })) + titles = {p.title for p in sess.query(Proposal).all()} + assert "Weekly crypto digest" not in titles # no keyword match in feed -> dropped + + +def test_scan_records_audit_for_each_proposal(cfg, sess): + from cayceops.models import AuditEntry + + config = _cfg(cfg) + scanner.scan(config, sess, fetch=_fake_fetch({ + "https://feeds.example.org/news.xml": RSS, + "https://pay.example.org/partners": PAGE, + })) + acts = [a.action for a in sess.query(AuditEntry).all()] + assert acts.count("propose") == 3 + assert all(a.actor == "daemon:scanner" for a in sess.query(AuditEntry).all()) + + +def test_scan_survives_source_error(cfg, sess): + def broken_fetch(url, cfg, timeout=20.0): + raise OSError("connection refused") + + config = _cfg(cfg) + stats = scanner.scan(config, sess, fetch=broken_fetch) + assert stats["proposed"] == 0 + assert len(stats["source_errors"]) == 2