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

feat: expand mailbox evidence scanner

Browse Source
This commit is contained in:
Bernd Worsch
2026-06-02 02:07:50 +02:00
parent 8532583182
commit 226c045397
16 changed files with 670 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
tests/test_scanner.py
View File

@@ -2,10 +2,12 @@ from __future__ import annotations
import tempfile
import unittest
from csv import DictReader
from pathlib import Path
from email_connect.config import AppConfig, MailboxConfig, ReportsConfig, ScanConfig, SourceConfig, StorageConfig
from email_connect.scanner import scan_mailbox
from email_connect.storage import StateStore
FIXTURES = Path(__file__).parent / "fixtures" / "mailbox"
@@ -24,13 +26,44 @@ class ScannerTests(unittest.TestCase):
)
first = scan_mailbox(config)
second = scan_mailbox(config)
full = scan_mailbox(config, full_rescan=True, report_only_new=True)
self.assertEqual(first.scan.messages_seen, 4)
self.assertEqual(first.scan.messages_new, 4)
self.assertGreaterEqual(first.scan.evidence_events_created, 4)
self.assertEqual(first.scan.messages_seen, 9)
self.assertEqual(first.scan.messages_new, 9)
self.assertGreaterEqual(first.scan.evidence_events_created, 9)
self.assertEqual(second.scan.messages_seen, 0)
self.assertEqual(second.scan.messages_new, 0)
self.assertEqual(second.scan.evidence_events_created, 0)
self.assertEqual(full.scan.messages_seen, 9)
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())
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)), [])
def test_scan_updates_endpoint_quality(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)),
)
scan_mailbox(config)
store = StateStore(config.storage.path)
try:
rows = {row["affected_email_address"]: row for row in store.endpoint_quality_rows()}
finally:
store.close()
self.assertEqual(rows["missing@example.com"]["reachability"], "unreachable")
self.assertEqual(rows["full@example.com"]["reachability"], "degraded")
self.assertEqual(rows["complained@example.com"]["suppression_state"], "suppressed")
self.assertEqual(rows["optout@example.com"]["suppression_state"], "opted_out")
if __name__ == "__main__":