cayce
a282aedb94
All 32 findings resolved: regex literals kept single-line with noqa (pattern IS the reviewed artifact), HTML rows string-concatenated for line length, walrus tuple in BIP-340 verify split into named generator, unused pytest import dropped. No behavior changes; 44/44 tests pass. Full-gate discipline restored: ruff exit 0.
183 lines
6.6 KiB
Python
183 lines
6.6 KiB
Python
"""maild tests — parsing, classification, threading, gated send. Invented data only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from cayceops.components.maild import classify as clf
|
|
from cayceops.components.maild import maild
|
|
from cayceops.models import (
|
|
AuditEntry,
|
|
DraftStatus,
|
|
MailClassification,
|
|
MailDirection,
|
|
MailMessage,
|
|
Proposal,
|
|
ProposalStatus,
|
|
)
|
|
|
|
OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
|
|
|
|
|
|
def _eml(
|
|
subject: str,
|
|
body: str,
|
|
in_reply_to: str | None = None,
|
|
msgid: str | None = None,
|
|
) -> bytes:
|
|
lines = [
|
|
"From: Partner Desk <partners@acme.example.org>",
|
|
"To: agent@cayce-co.xyz",
|
|
f"Subject: {subject}",
|
|
f"Message-ID: {msgid or '<acme-reply-1@acme.example.org>'}",
|
|
]
|
|
if in_reply_to:
|
|
lines.append(f"In-Reply-To: {in_reply_to}")
|
|
lines += ["MIME-Version: 1.0", "Content-Type: text/plain; charset=utf-8", "", body]
|
|
return "\r\n".join(lines).encode()
|
|
|
|
|
|
def _proposal_with_sent_draft(sess) -> Proposal:
|
|
p = Proposal(
|
|
kind="partnership",
|
|
title="Acme hotel affiliate program",
|
|
status=ProposalStatus.APPROVED,
|
|
)
|
|
sess.add(p)
|
|
sess.flush()
|
|
sess.add(
|
|
MailMessage(
|
|
direction=MailDirection.OUTBOUND,
|
|
message_id=OUTBOUND_MSGID,
|
|
to_addr="partners@acme.example.org",
|
|
subject="[cayce] partnership inquiry",
|
|
proposal_id=p.id,
|
|
draft_status=DraftStatus.SENT,
|
|
)
|
|
)
|
|
sess.flush()
|
|
return p
|
|
|
|
|
|
# --- classifier ---------------------------------------------------------------
|
|
|
|
|
|
def test_rules_classification(cfg):
|
|
config = cfg()
|
|
ok = clf.classify("Your application was approved, welcome to the program", config)
|
|
no = clf.classify("Unfortunately we cannot accept new partners", config)
|
|
assert ok[0] is MailClassification.APPROVAL
|
|
assert no[0] is MailClassification.REJECTION
|
|
q = clf.classify("Could you provide your traffic numbers?", config)
|
|
assert q[0] is MailClassification.QUESTION
|
|
kind, confidence = clf.classify("grocery list", config)
|
|
assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
|
|
|
|
|
|
def test_low_confidence_becomes_needs_review(cfg):
|
|
config = cfg(CAYCE_OPS_LLM_REVIEW_THRESHOLD="0.95")
|
|
kind, confidence = clf.classify("Your application was approved, welcome", config)
|
|
assert kind is MailClassification.NEEDS_REVIEW
|
|
assert confidence < 0.95
|
|
|
|
|
|
# --- inbound pass ---------------------------------------------------------------
|
|
|
|
|
|
def test_poll_threads_approval_and_drafts_reply(cfg, sess):
|
|
p = _proposal_with_sent_draft(sess)
|
|
raw = _eml(
|
|
"Re: [cayce] partnership inquiry",
|
|
"Approved! Welcome to the Acme partner program.",
|
|
in_reply_to=OUTBOUND_MSGID,
|
|
)
|
|
|
|
stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
|
|
assert stats["processed"] == 1 and stats["drafted"] == 1
|
|
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert inbound.classification is MailClassification.APPROVAL
|
|
assert inbound.proposal_id == p.id
|
|
assert inbound.draft_status is DraftStatus.DRAFTED
|
|
assert inbound.draft_reply and "approval" in inbound.draft_reply.lower()
|
|
|
|
assert p.detail["merchant_status"] == "approval"
|
|
assert p.status is ProposalStatus.APPROVED # lifecycle untouched by maild
|
|
|
|
|
|
def test_poll_idempotent_on_message_id(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: hi", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
|
|
def fetcher(cfg):
|
|
return [raw]
|
|
|
|
assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1
|
|
assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1 # noqa: E501
|
|
|
|
|
|
def test_unthreaded_approval_needs_review(cfg, sess):
|
|
raw = _eml("Great news", "Approved! Welcome aboard.") # no In-Reply-To
|
|
stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert inbound.classification is MailClassification.NEEDS_REVIEW
|
|
assert stats["needs_review"] == 1
|
|
assert inbound.draft_status is DraftStatus.NONE # no draft for review items
|
|
|
|
|
|
def test_question_gets_clarification_draft(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Could you provide your monthly traffic?", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert inbound.classification is MailClassification.QUESTION
|
|
assert inbound.draft_status is DraftStatus.DRAFTED
|
|
assert "traffic" in inbound.draft_reply
|
|
|
|
|
|
# --- human gate -------------------------------------------------------------------
|
|
|
|
|
|
def test_send_blocked_without_approval(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub)
|
|
|
|
sent = []
|
|
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
|
|
stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
|
|
assert stats["sent"] == 0 and sent == []
|
|
row = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert row.draft_status is DraftStatus.DRAFTED
|
|
|
|
|
|
def test_send_requires_explicit_human_approval(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml(
|
|
"Re: inquiry",
|
|
"Approved, welcome to the program.",
|
|
in_reply_to=OUTBOUND_MSGID,
|
|
)
|
|
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub)
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
|
|
assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
|
|
|
|
sent = []
|
|
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
|
|
stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
|
|
assert stats["sent"] == 1
|
|
assert inbound.draft_status is DraftStatus.SENT
|
|
assert sent[0][1] == "partners@acme.example.org"
|
|
|
|
row = sess.query(AuditEntry).filter_by(action="send_draft").one()
|
|
assert "gustavo" in str(row.detail)
|
|
|
|
|
|
def test_double_approve_is_rejected(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub)
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
|
|
assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
|
|
assert maild._approve_draft(inbound.id, "gustavo", sess) == 1
|