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.mailbox import MaildirMailboxSource, source_for_config from email_connect.scanner import scan_mailbox FIXTURES = Path(__file__).parent / "fixtures" / "mailbox" def build_maildir(root: Path, *, subdir: str = "new", start: int = 1749000000) -> Path: """Materialize the .eml fixtures as a Maildir tree with realistic names.""" maildir = root / "Maildir" for name in ("new", "cur", "tmp"): (maildir / name).mkdir(parents=True, exist_ok=True) for offset, source in enumerate(sorted(FIXTURES.glob("*.eml"))): filename = f"{start + offset}.M{offset}P100.harness" if subdir == "cur": filename += ":2,S" (maildir / subdir / filename).write_bytes(source.read_bytes()) return maildir def maildir_config(root: Path, maildir: Path) -> AppConfig: return AppConfig( mailbox=MailboxConfig(id="test-maildir", protocol="maildir"), scan=ScanConfig(), storage=StorageConfig(path=str(root / "state.sqlite")), reports=ReportsConfig(output_dir=str(root / "reports")), source=SourceConfig(maildir_dir=str(maildir)), ) def report_events(path: Path) -> set[tuple[str, str]]: with path.open(newline="", encoding="utf-8") as fh: return {(row["normalized_event_type"], row["affected_email_address"]) for row in DictReader(fh)} class MaildirSourceTests(unittest.TestCase): def test_maildir_scan_matches_fixture_scan(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) maildir_result = scan_mailbox(maildir_config(root, maildir)) fixture_root = root / "fixture-run" fixture_config = AppConfig( mailbox=MailboxConfig(id="test-mailbox", protocol="fixture"), scan=ScanConfig(), storage=StorageConfig(path=str(fixture_root / "state.sqlite")), reports=ReportsConfig(output_dir=str(fixture_root / "reports")), source=SourceConfig(fixture_dir=str(FIXTURES)), ) fixture_result = scan_mailbox(fixture_config) self.assertEqual(maildir_result.scan.messages_seen, fixture_result.scan.messages_seen) self.assertEqual(maildir_result.scan.messages_parsed, fixture_result.scan.messages_parsed) self.assertEqual( maildir_result.scan.evidence_events_created, fixture_result.scan.evidence_events_created, ) self.assertEqual( report_events(maildir_result.report_path), report_events(fixture_result.report_path), ) def test_incremental_cursor_and_full_rescan(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) config = maildir_config(root, maildir) first = scan_mailbox(config) second = scan_mailbox(config) self.assertEqual(first.scan.messages_new, 11) self.assertEqual(second.scan.messages_seen, 0) self.assertEqual(second.scan.evidence_events_created, 0) # A later delivery is picked up by the cursor without a rescan. (maildir / "new" / "1749000900.M99P100.harness").write_bytes( (FIXTURES / "hard_bounce.eml").read_bytes() ) third = scan_mailbox(config) self.assertEqual(third.scan.messages_seen, 1) self.assertEqual(third.scan.messages_new, 1) full = scan_mailbox(config, full_rescan=True) self.assertEqual(full.scan.messages_seen, 12) self.assertEqual(full.scan.messages_new, 0) self.assertEqual(full.scan.evidence_events_created, 0) def test_message_identity_survives_move_from_new_to_cur(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) config = maildir_config(root, maildir) scan_mailbox(config) for path in (maildir / "new").iterdir(): path.rename(maildir / "cur" / f"{path.name}:2,S") rescan = scan_mailbox(config, full_rescan=True) self.assertEqual(rescan.scan.messages_seen, 11) self.assertEqual(rescan.scan.messages_new, 0) self.assertEqual(rescan.scan.evidence_events_created, 0) def test_raw_message_ref_uses_maildir_scheme(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) messages = list( MaildirMailboxSource(maildir).iter_messages( max_messages=1, since_uid=None, full_rescan=False, include_seen=True, since=None, ) ) self.assertEqual(len(messages), 1) self.assertTrue(messages[0].raw_message_ref.startswith(f"maildir://{maildir}/new/")) self.assertIsNone(messages[0].imap_uid) self.assertEqual(messages[0].dedup_uid, messages[0].source_uid) self.assertNotIn(":", messages[0].source_uid) def test_include_seen_false_skips_flagged_messages(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root, subdir="cur") source = MaildirMailboxSource(maildir) all_messages = list( source.iter_messages( max_messages=0, since_uid=None, full_rescan=False, include_seen=True, since=None ) ) unseen_only = list( source.iter_messages( max_messages=0, since_uid=None, full_rescan=False, include_seen=False, since=None ) ) self.assertEqual(len(all_messages), 11) self.assertEqual(unseen_only, []) def test_mark_seen_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) source = MaildirMailboxSource(maildir, mark_seen=True) with self.assertRaises(ValueError): list( source.iter_messages( max_messages=0, since_uid=None, full_rescan=False, include_seen=True, since=None ) ) def test_missing_maildir_directory_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as tmp: source = MaildirMailboxSource(Path(tmp) / "absent") with self.assertRaises(ValueError): list( source.iter_messages( max_messages=0, since_uid=None, full_rescan=False, include_seen=True, since=None ) ) def test_source_for_config_requires_maildir_dir(self) -> None: config = AppConfig( mailbox=MailboxConfig(id="test-maildir", protocol="maildir"), scan=ScanConfig(), storage=StorageConfig(), reports=ReportsConfig(), source=SourceConfig(), ) with self.assertRaises(ValueError): source_for_config(config) def test_source_for_config_returns_maildir_source(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) maildir = build_maildir(root) source = source_for_config(maildir_config(root, maildir)) self.assertIsInstance(source, MaildirMailboxSource) if __name__ == "__main__": unittest.main()