coulomb/email-connect
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add expected recipient reporting

Browse Source
This commit is contained in:
Bernd Worsch
2026-06-02 03:07:13 +02:00
parent 5ea6c738d2
commit b7591f531b
17 changed files with 629 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
tests/test_scanner.py
View File

@@ -11,6 +11,7 @@ from email_connect.storage import StateStore
FIXTURES = Path(__file__).parent / "fixtures" / "mailbox"
RECIPIENT_FIXTURES = Path(__file__).parent / "fixtures"
class ScannerTests(unittest.TestCase):
@@ -38,6 +39,10 @@ class ScannerTests(unittest.TestCase):
self.assertEqual(full.scan.messages_new, 0)
self.assertEqual(full.scan.evidence_events_created, 0)
self.assertTrue(first.report_path and first.report_path.exists())
with first.report_path.open(newline="", encoding="utf-8") as fh:
first_rows = list(DictReader(fh))
self.assertTrue(first_rows)
self.assertTrue(all(row["known_recipient"] == "false" for row in first_rows))
self.assertTrue(full.report_path and full.report_path.exists())
with full.report_path.open(newline="", encoding="utf-8") as fh:
self.assertEqual(list(DictReader(fh)), [])
@@ -65,6 +70,110 @@ class ScannerTests(unittest.TestCase):
self.assertEqual(rows["complained@example.com"]["suppression_state"], "suppressed")
self.assertEqual(rows["optout@example.com"]["suppression_state"], "opted_out")
def test_expected_recipients_sort_first_and_get_no_evidence_rows(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
config = AppConfig(
mailbox=MailboxConfig(id="test-mailbox", protocol="fixture"),
scan=ScanConfig(),
storage=StorageConfig(path=str(root / "state.sqlite")),
reports=ReportsConfig(output_dir=str(root / "reports")),
source=SourceConfig(fixture_dir=str(FIXTURES)),
)
result = scan_mailbox(
config,
full_rescan=True,
expected_recipients_path=str(RECIPIENT_FIXTURES / "expected_recipients.txt"),
)
self.assertEqual(len(result.warnings), 1)
self.assertTrue(result.report_path and result.report_path.exists())
with result.report_path.open(newline="", encoding="utf-8") as fh:
rows = list(DictReader(fh))
self.assertGreater(len(rows), 2)
known_flags = [row["known_recipient"] for row in rows]
self.assertEqual(known_flags, sorted(known_flags, reverse=True))
missing_rows = [row for row in rows if row["affected_email_address"] == "missing@example.com"]
self.assertTrue(missing_rows)
self.assertTrue(all(row["known_recipient"] == "true" for row in missing_rows))
absent_rows = [row for row in rows if row["affected_email_address"] == "absent@example.com"]
self.assertEqual(len(absent_rows), 1)
self.assertEqual(absent_rows[0]["normalized_event_type"], "diagnostic.expected_recipient.no_evidence")
self.assertEqual(absent_rows[0]["assessment_category"], "undef")
self.assertEqual(absent_rows[0]["assessment_subclass"], "undef.no_signal")
self.assertEqual(absent_rows[0]["evidence_strength"], "none")
store = StateStore(config.storage.path)
try:
quality_addresses = {row["affected_email_address"] for row in store.endpoint_quality_rows()}
finally:
store.close()
self.assertNotIn("absent@example.com", quality_addresses)
def test_csv_expected_recipients_are_supported(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
config = AppConfig(
mailbox=MailboxConfig(id="test-mailbox", protocol="fixture"),
scan=ScanConfig(),
storage=StorageConfig(path=str(root / "state.sqlite")),
reports=ReportsConfig(output_dir=str(root / "reports")),
source=SourceConfig(fixture_dir=str(FIXTURES)),
)
result = scan_mailbox(
config,
full_rescan=True,
expected_recipients_path=str(RECIPIENT_FIXTURES / "expected_recipients.csv"),
expected_recipient_column="email",
)
self.assertEqual(len(result.warnings), 1)
self.assertTrue(result.report_path and result.report_path.exists())
with result.report_path.open(newline="", encoding="utf-8") as fh:
rows = list(DictReader(fh))
optout_rows = [row for row in rows if row["affected_email_address"] == "optout@example.com"]
self.assertTrue(optout_rows)
self.assertTrue(all(row["known_recipient"] == "true" for row in optout_rows))
csv_absent = [row for row in rows if row["affected_email_address"] == "csv-absent@example.com"]
self.assertEqual(csv_absent[0]["normalized_event_type"], "diagnostic.expected_recipient.no_evidence")
def test_datetime_range_excludes_messages_outside_range(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
expected_path = root / "expected.txt"
expected_path.write_text("complained@example.com\nmissing@example.com\n", encoding="utf-8")
config = AppConfig(
mailbox=MailboxConfig(id="test-mailbox", protocol="fixture"),
scan=ScanConfig(),
storage=StorageConfig(path=str(root / "state.sqlite")),
reports=ReportsConfig(output_dir=str(root / "reports")),
source=SourceConfig(fixture_dir=str(FIXTURES)),
)
result = scan_mailbox(
config,
full_rescan=True,
expected_recipients_path=str(expected_path),
range_from="2026-06-02T10:04:00Z",
range_to="2026-06-02T10:04:00Z",
)
self.assertEqual(result.scan.messages_seen, 11)
self.assertEqual(result.scan.messages_parsed, 1)
self.assertTrue(result.report_path and result.report_path.exists())
with result.report_path.open(newline="", encoding="utf-8") as fh:
rows = list(DictReader(fh))
self.assertEqual({row["affected_email_address"] for row in rows}, {"complained@example.com", "missing@example.com"})
complaint = [row for row in rows if row["affected_email_address"] == "complained@example.com"][0]
missing = [row for row in rows if row["affected_email_address"] == "missing@example.com"][0]
self.assertEqual(complaint["normalized_event_type"], "notification.channel.complaint_received")
self.assertEqual(missing["normalized_event_type"], "diagnostic.expected_recipient.no_evidence")
self.assertEqual(result.scan.range_start.isoformat(), "2026-06-02T10:04:00+00:00")
self.assertEqual(result.scan.range_end.isoformat(), "2026-06-02T10:04:00+00:00")
if __name__ == "__main__":
unittest.main()