cayce
0d24309433
ci / lint-test-leaks (pull_request) Canceled after 0s
Background: CAYCE_OPS_OPERATOR_PUBKEY was set to Cayce nostr pubkey — Cayce was approving itself. Signature was technically valid but semantically null: human approval signed by the agent is not human approval. - signing.py: validate_no_self_approval() rejects the operator pubkey when it matches (a) the pubkey derived from any private key in the config, or (b) any pubkey in CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS deny-list. - executor/maild/publisher _signing() now calls the validation, so every daemon and every CLI approve-* path refuses to operate (D16). - tests: 6 new (agent-key refusal, human key passes, empty passes, explicit denylist, end-to-end daemon refusal, derivation roundtrip). - docs/DECISIONS.md: D16. The operator signature is now produced OUTSIDE the VPS. CAYCE_OPS_OPERATOR_PUBKEY must stay empty until the external signer exists — daemons refuse, which is correct. The Cayce nsec itself is NOT touched (still the agent identity for publishing and anchoring).
533 lines
20 KiB
Python
533 lines
20 KiB
Python
"""maild tests — parsing, classification, threading, gated send. Invented data only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from coincurve import PrivateKey
|
|
from coincurve import PublicKey as _PK
|
|
|
|
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,
|
|
)
|
|
|
|
# test operator keypair (matches conftest default pubkey; invented)
|
|
OPERATOR_SECRET = bytes.fromhex("22" * 32)
|
|
OPERATOR_PUBKEY = _PK.from_secret(OPERATOR_SECRET).format().hex()[2:]
|
|
|
|
OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
|
|
|
|
# the signing gate reads env at runtime — make sure every path in this module
|
|
# (fixtures included) sees a signed deployment
|
|
os.environ.setdefault("CAYCE_OPS_DEPLOYMENT_ID", "itest")
|
|
os.environ.setdefault("CAYCE_OPS_OPERATOR_PUBKEY", OPERATOR_PUBKEY)
|
|
from cayceops.config import load as _load # noqa: E402 (env set above)
|
|
|
|
STUB_CFG = _load() # a real, signed Config (replaces the old SimpleNamespace)
|
|
|
|
|
|
|
|
def _eml(subject, body, in_reply_to=None, msgid=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 _sign(msg: bytes, secret: bytes = OPERATOR_SECRET) -> str:
|
|
|
|
return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()
|
|
|
|
|
|
def _approve_signed(sess, mail_id: int) -> str:
|
|
"""Approve with a valid operator signature over the row's current content."""
|
|
from cayceops.signing import approval_message, mail_content_hash
|
|
|
|
record = sess.get(MailMessage, mail_id)
|
|
to_addr = record.from_addr if record.direction == MailDirection.INBOUND else record.to_addr
|
|
chash = mail_content_hash(to_addr, record.subject, record.draft_reply)
|
|
return _sign(approval_message("mail", "itest", mail_id, chash))
|
|
|
|
|
|
def _proposal_with_sent_draft(sess) -> Proposal:
|
|
"""A proposal whose first-contact mail went through the REAL path:
|
|
compose -> approve (signed) -> send (creating a true OUTBOUND row)."""
|
|
p = Proposal(
|
|
kind="partnership",
|
|
title="Acme hotel affiliate program",
|
|
status=ProposalStatus.APPROVED,
|
|
)
|
|
sess.add(p)
|
|
sess.flush()
|
|
mail_id = maild._compose(
|
|
p.id, "partners@acme.example.org", "[cayce] partnership inquiry", "Hello!", sess
|
|
)
|
|
assert (
|
|
maild._approve_draft(mail_id, "operator", sess, signature=_approve_signed(sess, mail_id))
|
|
== 0
|
|
)
|
|
assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
|
|
out = (
|
|
sess.query(MailMessage)
|
|
.filter_by(direction=MailDirection.OUTBOUND, proposal_id=p.id)
|
|
.one()
|
|
)
|
|
out.message_id = OUTBOUND_MSGID # deterministic id for the fixtures
|
|
sess.flush()
|
|
return p
|
|
|
|
|
|
def _fetcher(raws: list[bytes]):
|
|
return lambda cfg: [(bytes(str(i), "ascii"), raw) for i, raw in enumerate(raws)]
|
|
|
|
|
|
# --- classifier ---------------------------------------------------------------
|
|
|
|
|
|
def test_rules_classification(cfg):
|
|
config = cfg()
|
|
assert (
|
|
clf.classify("Your application was approved, welcome to the program", config)[0]
|
|
is MailClassification.APPROVAL
|
|
)
|
|
# negation present -> approval is vetoed to needs_review (round-2 rule;
|
|
# the human decides between rejection and other)
|
|
assert (
|
|
clf.classify("Unfortunately we cannot accept new partners", config)[0]
|
|
is MailClassification.NEEDS_REVIEW
|
|
)
|
|
# plain rejection (no approval word involved) still classifies directly
|
|
assert (
|
|
clf.classify("We regret to inform you that your application was declined", config)[0]
|
|
is MailClassification.REJECTION
|
|
)
|
|
assert (
|
|
clf.classify("Could you provide your traffic numbers?", config)[0]
|
|
is MailClassification.QUESTION
|
|
)
|
|
kind, _ = clf.classify("grocery list", config)
|
|
assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
|
|
|
|
|
|
def test_negated_approvals_never_classify_approval(cfg):
|
|
"""Round-2 point 2: any negation vetoes `approval` -> needs_review."""
|
|
config = cfg()
|
|
cases = [
|
|
"We cannot currently accept new affiliates",
|
|
"We will never accept crypto businesses",
|
|
"Approval was not granted for your site",
|
|
"We are no longer accepting applications",
|
|
"Your application hasn't been approved",
|
|
"We have decided not to move forward with your application",
|
|
"We cannot accept your application",
|
|
"We are not accepting new partners at this time",
|
|
"Your application was not approved",
|
|
"We can't approve this request",
|
|
"We do not accept external affiliates",
|
|
]
|
|
for text in cases:
|
|
kind, _ = clf.classify(text, config)
|
|
assert kind is not MailClassification.APPROVAL, f"{text!r} -> {kind}"
|
|
|
|
|
|
def test_llm_prompt_format_survives_json_braces():
|
|
"""Critical finding 4: .format() raised KeyError on the JSON sample."""
|
|
formatted = clf._LLM_PROMPT.format(body="x")
|
|
assert '{"classification"' in formatted # literal braces survive
|
|
|
|
|
|
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, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
assert stats["processed"] == 1 and stats["drafted"] == 1
|
|
assert len(nums) == 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)
|
|
|
|
first, second = maild._poll(cfg(), sess, fetcher=_fetcher([raw])), maild._poll(
|
|
cfg(), sess, fetcher=_fetcher([raw])
|
|
)
|
|
assert first[0]["processed"] == 1
|
|
assert second[0]["skipped"] == 1
|
|
|
|
|
|
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=_fetcher([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=_fetcher([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=_fetcher([raw]))
|
|
|
|
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 == []
|
|
assert stats["blocked"] + stats["missing"] >= 0 # no crash paths
|
|
assert config.mail_from == "agent@cayce-co.xyz"
|
|
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=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
original_id = inbound.message_id
|
|
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
|
|
sent = []
|
|
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
|
|
|
|
def sender(cfg, to, subject, body, message_id, in_reply_to=None):
|
|
sent.append((to, subject, body, message_id, in_reply_to))
|
|
|
|
stats = maild._send_approved(config, sess, sender=sender)
|
|
assert stats["sent"] == 1
|
|
assert inbound.draft_status is DraftStatus.SENT
|
|
to, subject, body, message_id, in_reply_to = sent[0]
|
|
assert to == "partners@acme.example.org"
|
|
assert in_reply_to == original_id # In-Reply-To threads to the inbound mail
|
|
|
|
# a real OUTBOUND row now exists (audit fix: they were never created)
|
|
out = (
|
|
sess.query(MailMessage)
|
|
.filter_by(direction=MailDirection.OUTBOUND, message_id=message_id)
|
|
.one()
|
|
)
|
|
assert out.in_reply_to == original_id
|
|
assert out.proposal_id == inbound.proposal_id
|
|
assert inbound.message_id == original_id # inbound id NOT overwritten
|
|
|
|
row = (
|
|
sess.query(AuditEntry)
|
|
.filter_by(action="send_draft")
|
|
.order_by(AuditEntry.id.desc())
|
|
.first()
|
|
)
|
|
assert row is not None and "operator" in str(row.detail)
|
|
|
|
|
|
def test_double_approve_is_rejected(cfg, sess):
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
assert maild._approve_draft(inbound.id, "operator", sess) == 1
|
|
|
|
|
|
# --- audit corrections ----------------------------------------------------------
|
|
|
|
|
|
def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
|
|
"""Critical finding 3: SMTP failure mid-batch rolled back the whole
|
|
transaction, re-sending already-delivered mails 5 minutes later."""
|
|
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
|
|
for i in range(3):
|
|
# unique first-contact thread per proposal (real compose -> send path)
|
|
p = Proposal(kind="partnership", title=f"Acme partner {i}")
|
|
sess.add(p)
|
|
sess.flush()
|
|
mail_id = maild._compose(
|
|
p.id, f"desk{i}@acme.example.org", f"[cayce] inquiry {i}", "Hello!", sess
|
|
)
|
|
assert (
|
|
maild._approve_draft(
|
|
mail_id, "operator", sess, signature=_approve_signed(sess, mail_id)
|
|
)
|
|
== 0
|
|
)
|
|
assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
|
|
sess.flush()
|
|
out = sess.get(MailMessage, mail_id)
|
|
raw = _eml(
|
|
f"Re: {i}",
|
|
"Approved, welcome to the program.",
|
|
in_reply_to=out.message_id,
|
|
msgid=f"<acme-{i}@acme.example.org>",
|
|
)
|
|
maild._poll(config, sess, fetcher=_fetcher([raw]))
|
|
inbound = (
|
|
sess.query(MailMessage)
|
|
.filter_by(direction=MailDirection.INBOUND, message_id=f"<acme-{i}@acme.example.org>")
|
|
.one()
|
|
)
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
sess.commit() # per-message factory sessions only see committed rows
|
|
|
|
calls = {"n": 0}
|
|
|
|
def flaky_smtp(cfg, to, subject, body, message_id, in_reply_to=None):
|
|
calls["n"] += 1
|
|
if calls["n"] == 3:
|
|
raise OSError("smtp connection lost")
|
|
|
|
import cayceops.db as appdb
|
|
|
|
stats = maild._send_approved(
|
|
config, sess, sender=flaky_smtp, factory=appdb.session_factory()
|
|
)
|
|
|
|
assert calls["n"] == 3
|
|
assert stats["sent"] == 2 and stats["error"] == 1 # no rollback of 1..2
|
|
still_approved = (
|
|
sess.query(MailMessage)
|
|
.filter_by(direction=MailDirection.INBOUND, draft_status=DraftStatus.APPROVED)
|
|
.count()
|
|
)
|
|
assert still_approved == 1 # only the failed one stays queued
|
|
|
|
|
|
def test_edited_draft_refuses_to_send(cfg, sess):
|
|
"""Medium finding: approved text was not bound to its content."""
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
|
|
inbound.draft_reply = "TOTALLY EDITED TEXT after approval" # tamper
|
|
|
|
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 stats["hash_mismatch"] == 1
|
|
assert sent == []
|
|
assert inbound.draft_status is DraftStatus.DRAFTED # needs fresh approval
|
|
action = sess.query(AuditEntry).filter_by(action="send_refused").one()
|
|
assert action.result == "error"
|
|
|
|
|
|
def test_compose_creates_first_contact_draft(cfg, sess):
|
|
"""Critical finding 2: there was no path for the first contact mail."""
|
|
p = Proposal(kind="partnership", title="New hotel partner")
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
mail_id = maild._compose(
|
|
p.id, "desk@hotel.example.org", "Partnership", "We would like to...", sess
|
|
)
|
|
assert mail_id > 0
|
|
out = sess.get(MailMessage, mail_id)
|
|
assert out.direction is MailDirection.OUTBOUND
|
|
assert out.draft_status is DraftStatus.DRAFTED
|
|
assert out.proposal_id == p.id
|
|
|
|
# approve + send -> an inbound reply can now thread to this proposal
|
|
assert (
|
|
maild._approve_draft(mail_id, "operator", sess, signature=_approve_signed(sess, mail_id))
|
|
== 0
|
|
)
|
|
stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None)
|
|
assert stats["sent"] == 1
|
|
sess.flush()
|
|
|
|
reply = _eml("Re: Partnership", "Approved, welcome!", in_reply_to=out.message_id)
|
|
assert maild._match_proposal(maild.parse_raw(reply), sess) is p
|
|
|
|
|
|
def test_mark_seen_returned_for_post_commit_processing(cfg, sess):
|
|
"""Medium finding: \\\\Seen was set before processing (crash lost mail)."""
|
|
raw = _eml("Re: hi", "Approved, welcome to the program.")
|
|
stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
assert stats["processed"] == 1
|
|
assert nums == [b"0"] # caller marks seen only after commit
|
|
|
|
|
|
# --- round-2 signature gate (mail) ------------------------------------------------
|
|
|
|
|
|
def _approved_reply(sess, cfg) -> MailMessage:
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
return sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
|
|
|
|
def test_unsigned_sql_approved_draft_is_refused(cfg, sess):
|
|
"""Round-2 required test: raw-SQL approved row (no signature) not sent."""
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
# raw-SQL style approval: status flips, no signature
|
|
sess.execute(
|
|
MailMessage.__table__.update()
|
|
.where(MailMessage.id == inbound.id)
|
|
.values(draft_status=DraftStatus.APPROVED)
|
|
)
|
|
sess.expire(inbound)
|
|
|
|
sent = []
|
|
stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a))
|
|
assert stats["sent"] == 0 and stats["hash_mismatch"] == 1
|
|
assert sent == []
|
|
assert inbound.draft_status is DraftStatus.DRAFTED # reverted
|
|
refused = sess.query(AuditEntry).filter_by(action="send_refused").one()
|
|
assert refused.result == "error"
|
|
|
|
|
|
def test_changed_to_addr_after_approval_blocks_send(cfg, sess):
|
|
"""Round-2 required test: to_addr edited after approval -> refused."""
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
|
|
inbound.from_addr = "attacker@evil.example.org" # redirect target
|
|
|
|
sent = []
|
|
stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a))
|
|
assert stats["sent"] == 0 and stats["hash_mismatch"] == 1
|
|
assert sent == []
|
|
assert inbound.draft_status is DraftStatus.DRAFTED
|
|
|
|
|
|
def test_foreign_deployment_signature_refused(cfg, sess):
|
|
"""Round-2 required test: a valid signature for another deployment_id (or
|
|
other content) is refused."""
|
|
from cayceops.signing import approval_message, mail_content_hash
|
|
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
chash = mail_content_hash(inbound.from_addr, inbound.subject, inbound.draft_reply)
|
|
|
|
# signature valid for ANOTHER deployment
|
|
other_dep = _sign(
|
|
OPERATOR_SECRET, approval_message("mail", "other-deploy", inbound.id, chash)
|
|
)
|
|
assert maild._approve_draft(inbound.id, "operator", sess, signature=other_dep) == 2
|
|
assert inbound.draft_status is DraftStatus.DRAFTED
|
|
|
|
# signature valid for OTHER CONTENT
|
|
other_content = _sign(
|
|
OPERATOR_SECRET, approval_message("mail", "itest", inbound.id, "00" * 32)
|
|
)
|
|
assert maild._approve_draft(inbound.id, "operator", sess, signature=other_content) == 2
|
|
assert inbound.draft_status is DraftStatus.DRAFTED
|
|
|
|
|
|
def test_send_failures_audited_and_exit_code(cfg, sess, monkeypatch):
|
|
"""Round-2 point 5: failures audited + surfaced; non-zero exit handled by CLI."""
|
|
import cayceops.db as appdb
|
|
|
|
_proposal_with_sent_draft(sess)
|
|
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
|
|
maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
|
|
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
|
|
assert (
|
|
maild._approve_draft(
|
|
inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
|
|
)
|
|
== 0
|
|
)
|
|
|
|
def boom(cfg, to, subject, body, message_id, in_reply_to=None):
|
|
raise OSError("relay down")
|
|
|
|
|
|
sess.commit() # per-message factory sessions only see committed rows
|
|
stats = maild._send_approved(
|
|
STUB_CFG, sess, sender=boom, factory=appdb.session_factory()
|
|
)
|
|
assert stats["error"] == 1 and stats["sent"] == 0
|
|
failed = sess.query(AuditEntry).filter_by(action="send_failed").one()
|
|
assert failed.result == "error"
|