generated from coulomb/repo-seed
71 lines
3.1 KiB
Python
71 lines
3.1 KiB
Python
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"
|
|
|
|
|
|
class ScannerTests(unittest.TestCase):
|
|
def test_scan_fixture_directory_writes_report_and_deduplicates(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)),
|
|
)
|
|
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, 11)
|
|
self.assertEqual(first.scan.messages_new, 11)
|
|
self.assertGreaterEqual(first.scan.evidence_events_created, 11)
|
|
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, 11)
|
|
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__":
|
|
unittest.main()
|