email-connect/tests/test_maildir.py

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()