"""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 = "" def _eml( subject: str, body: str, in_reply_to: str | None = None, msgid: str | None = None, ) -> bytes: lines = [ "From: Partner Desk ", "To: agent@cayce-co.xyz", f"Subject: {subject}", f"Message-ID: {msgid or ''}", ] 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