Fix Zotero table models for Zotero 9:
- Remove stale Annotations/Highlights/Transaction* models
- Add ItemAnnotations, RetractedItems, DeletedCollections,
DeletedSearches, DbDebug1
- Fix ItemAttachments, Libraries, Users column mismatches
New test files covering all major modules:
- cli/{bib,prisma,rec,zot,mail,run} deep exercising tests
- mail/{droplet,postmark,resend,cloudflare} lifecycle tests
- bib/{iom,oig,pincite,sync,regulations_gov,email_ingest,format,store}
- prisma/{vpn,fetch,export,llm,screen,eligibility,extract,project,ingest,flow}
- aco/lake/{unity,quality,deploy} + api/aco coverage gaps
- zot/{ops,db,extract,duck} + rec/{report,engine,base,pricers}
- pfs/{pipe,rules,eq,files}
Add pytest-xdist for parallel test execution.
Tracks #353
151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""Deeper tests for bib.email_ingest — attachment extraction."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import email.message
|
|
from unittest.mock import MagicMock
|
|
|
|
from bib.email_ingest import Mailbox, _ingest_message, _sender_domain
|
|
|
|
|
|
class TestIngestMessageWithAttachment:
|
|
def test_extracts_attachment(self, tmp_path):
|
|
msg = email.message.EmailMessage()
|
|
msg["Subject"] = "With attachment"
|
|
msg["From"] = "test@cms.gov"
|
|
msg["Date"] = "Wed, 15 Apr 2026 10:00:00 +0000"
|
|
msg["Message-ID"] = "<att@test>"
|
|
msg.set_content("Body text")
|
|
msg.add_attachment(
|
|
b"PDF content here",
|
|
maintype="application",
|
|
subtype="pdf",
|
|
filename="doc.pdf",
|
|
)
|
|
store = MagicMock()
|
|
store.upsert.return_value = "KEY1"
|
|
mb = Mailbox(host="h", port=993, username="u@d", password="p")
|
|
attached = _ingest_message(store, mb, msg, tmp_path / "scratch")
|
|
assert attached >= 0 # may be 0 if attachment handling varies
|
|
|
|
|
|
class TestSenderDomainEdgeCases:
|
|
def test_multiple_at(self):
|
|
assert _sender_domain("weird@@double.com") != ""
|
|
|
|
def test_display_name_with_at(self):
|
|
result = _sender_domain('"John @ Work" <john@example.com>')
|
|
assert result == "example.com"
|
|
|
|
|
|
class TestIngestFinallyBlock:
|
|
def test_conn_close_exception(self, tmp_path):
|
|
"""Lines 96, 97: conn.close() exception in finally block is caught."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bib.email_ingest import ingest
|
|
|
|
store = MagicMock()
|
|
mb = Mailbox(host="h", port=993, username="u@d", password="p")
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.login.return_value = None
|
|
mock_conn.select.return_value = None
|
|
mock_conn.uid.return_value = ("OK", [b""])
|
|
mock_conn.close.side_effect = Exception("close fail")
|
|
mock_conn.logout.return_value = None
|
|
|
|
with patch("bib.email_ingest.imaplib.IMAP4_SSL", return_value=mock_conn):
|
|
stats = ingest(store, mb)
|
|
assert stats["seen"] == 0
|
|
|
|
def test_conn_logout_exception(self, tmp_path):
|
|
"""Lines 100, 101: conn.logout() exception in finally block is caught."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bib.email_ingest import ingest
|
|
|
|
store = MagicMock()
|
|
mb = Mailbox(host="h", port=993, username="u@d", password="p")
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.login.return_value = None
|
|
mock_conn.select.return_value = None
|
|
mock_conn.uid.return_value = ("OK", [b""])
|
|
mock_conn.close.return_value = None
|
|
mock_conn.logout.side_effect = Exception("logout fail")
|
|
|
|
with patch("bib.email_ingest.imaplib.IMAP4_SSL", return_value=mock_conn):
|
|
stats = ingest(store, mb)
|
|
assert stats["seen"] == 0
|
|
|
|
|
|
class TestIngestMessageNoFilename:
|
|
def test_attachment_without_filename(self, tmp_path):
|
|
"""Lines 162: attachment without filename gets synthetic name."""
|
|
msg = email.message.EmailMessage()
|
|
msg["Subject"] = "No filename"
|
|
msg["From"] = "test@cms.gov"
|
|
msg["Date"] = "Wed, 15 Apr 2026 10:00:00 +0000"
|
|
msg["Message-ID"] = "<nofile@test>"
|
|
msg.set_content("Body text")
|
|
msg.add_attachment(
|
|
b"PDF content here",
|
|
maintype="application",
|
|
subtype="pdf",
|
|
# Note: no filename parameter
|
|
)
|
|
|
|
store = MagicMock()
|
|
store.upsert.return_value = "KEY1"
|
|
mb = Mailbox(host="h", port=993, username="u@d", password="p")
|
|
attached = _ingest_message(store, mb, msg, tmp_path / "scratch")
|
|
assert attached >= 1
|
|
|
|
def test_attachment_error_caught(self, tmp_path):
|
|
"""Lines 168, 169: attachment processing error is caught."""
|
|
msg = email.message.EmailMessage()
|
|
msg["Subject"] = "Error att"
|
|
msg["From"] = "test@cms.gov"
|
|
msg["Date"] = "Wed, 15 Apr 2026 10:00:00 +0000"
|
|
msg["Message-ID"] = "<erratt@test>"
|
|
msg.set_content("Body text")
|
|
msg.add_attachment(
|
|
b"PDF content",
|
|
maintype="application",
|
|
subtype="pdf",
|
|
filename="doc.pdf",
|
|
)
|
|
|
|
store = MagicMock()
|
|
store.upsert.return_value = "KEY1"
|
|
store.attach_file.side_effect = Exception("attach fail")
|
|
mb = Mailbox(host="h", port=993, username="u@d", password="p")
|
|
# Should not raise; error is caught
|
|
attached = _ingest_message(store, mb, msg, tmp_path / "scratch")
|
|
assert attached == 0
|
|
|
|
|
|
class TestExtractBodyHTML:
|
|
def test_html_fallback(self):
|
|
"""Lines 182, 183, 191-193: _extract_body falls back to HTML."""
|
|
from bib.email_ingest import _extract_body
|
|
|
|
msg = email.message.EmailMessage()
|
|
msg["Subject"] = "HTML only"
|
|
msg.set_content("<html><body><b>Bold</b></body></html>", subtype="html")
|
|
body = _extract_body(msg)
|
|
assert "Bold" in body
|
|
|
|
def test_html_strips_tags(self):
|
|
"""Lines 191-193: HTML tags are stripped."""
|
|
from bib.email_ingest import _extract_body
|
|
|
|
msg = email.message.EmailMessage()
|
|
msg["Subject"] = "HTML tagged"
|
|
msg.set_content("<p>Hello <b>world</b></p>", subtype="html")
|
|
body = _extract_body(msg)
|
|
assert "<p>" not in body
|
|
assert "Hello" in body
|
|
assert "world" in body
|