| import unittest |
| from unittest.mock import AsyncMock, patch |
|
|
| from app.models import ( |
| DNSResult, |
| DisposableResult, |
| SMTPResult, |
| SyntaxResult, |
| ValidationStatus, |
| ) |
| from app.validator.engine import EmailValidationEngine |
|
|
|
|
| def _dns_ok(host="mx.example.test"): |
| return DNSResult( |
| domain_exists=True, |
| has_mx_records=True, |
| mx_records=[{"host": host, "priority": 10}], |
| ) |
|
|
|
|
| def _disposable_clean(): |
| return DisposableResult( |
| is_disposable=False, is_free_provider=False, is_role_based=False |
| ) |
|
|
|
|
| class NullCache: |
| """Hermetic no-op cache. |
| |
| The engine defaults to a process-global cache for cross-run reuse; tests |
| must inject this so one test's cached DNS/MX cannot leak into the next. |
| """ |
|
|
| async def get_dns(self, domain): |
| return None |
|
|
| async def put_dns(self, domain, result): |
| return None |
|
|
| async def get_verdict(self, email, deep): |
| return None |
|
|
| async def put_verdict(self, email, deep, smtp, catch_all): |
| return None |
|
|
|
|
| class FakeBatchProvider: |
| """Provider exposing the bulk capability the engine should exploit. |
| |
| `verify_many` is the whole point: one call for N addresses lets the SMTP |
| engine reuse a single session per domain instead of one TCP handshake each. |
| """ |
|
|
| def __init__(self, results=None, catch_all=None): |
| self._results = results or {} |
| self._catch_all = catch_all or {} |
| self.batch_calls: list[list[tuple[str, str]]] = [] |
| self.single_calls = 0 |
|
|
| async def verify_many(self, pairs): |
| self.batch_calls.append(list(pairs)) |
| return dict(self._results), dict(self._catch_all) |
|
|
| async def verify(self, email, mx_host): |
| self.single_calls += 1 |
| return self._results.get(email, SMTPResult(verified=None, error="n/a")) |
|
|
| async def catch_all(self, domain, mx_host): |
| value = self._catch_all.get(domain) |
| return (value, 0.9) if value is not None else (None, 0.0) |
|
|
|
|
| class EngineBatchTests(unittest.IsolatedAsyncioTestCase): |
| """Bulk must go through ONE batched mailbox stage per run, not one |
| connection per address (the throughput requirement).""" |
|
|
| async def test_batch_resolves_dns_once_per_domain(self): |
| provider = FakeBatchProvider( |
| results={ |
| "a@corp.test": SMTPResult(verified=True, smtp_code=250), |
| "b@corp.test": SMTPResult(verified=False, smtp_code=550, |
| smtp_message="550 mailbox does not exist"), |
| "c@other.test": SMTPResult(verified=True, smtp_code=250), |
| }, |
| catch_all={"corp.test": False, "other.test": False}, |
| ) |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch( |
| "app.validator.engine.check_dns", |
| new=AsyncMock(return_value=_dns_ok()), |
| ) as dns_mock, |
| patch( |
| "app.validator.engine.check_disposable", |
| return_value=_disposable_clean(), |
| ), |
| ): |
| results = await engine.validate_batch( |
| ["a@corp.test", "b@corp.test", "c@other.test"], deep=True |
| ) |
|
|
| |
| self.assertEqual(dns_mock.await_count, 2) |
| self.assertEqual(len(results), 3) |
|
|
| async def test_batch_calls_provider_verify_many_once(self): |
| provider = FakeBatchProvider( |
| results={ |
| "a@corp.test": SMTPResult(verified=True, smtp_code=250), |
| "b@corp.test": SMTPResult(verified=True, smtp_code=250), |
| }, |
| catch_all={"corp.test": False}, |
| ) |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable_clean()), |
| ): |
| await engine.validate_batch(["a@corp.test", "b@corp.test"], deep=True) |
|
|
| self.assertEqual(len(provider.batch_calls), 1) |
| self.assertEqual( |
| sorted(provider.batch_calls[0]), |
| [("a@corp.test", "mx.example.test"), ("b@corp.test", "mx.example.test")], |
| ) |
| |
| self.assertEqual(provider.single_calls, 0) |
|
|
| async def test_batch_results_carry_provenance(self): |
| provider = FakeBatchProvider( |
| results={"a@corp.test": SMTPResult(verified=True, smtp_code=250)}, |
| catch_all={"corp.test": False}, |
| ) |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| r = results[0] |
| self.assertEqual( |
| r.checks_run, |
| ["syntax", "dns", "mx", "disposable", "catch_all", "smtp", "scoring"], |
| ) |
| self.assertTrue(r.run_id) |
| self.assertEqual(r.status, ValidationStatus.VALID) |
|
|
| async def test_batch_catch_all_rows_are_risky_not_valid(self): |
| provider = FakeBatchProvider( |
| results={ |
| "x@catchall.test": SMTPResult( |
| verified=None, |
| error="catch_all: catchall.test accepts every recipient", |
| ) |
| }, |
| catch_all={"catchall.test": True}, |
| ) |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable_clean()), |
| ): |
| results = await engine.validate_batch(["x@catchall.test"], deep=True) |
|
|
| self.assertEqual(results[0].status, ValidationStatus.RISKY) |
| self.assertEqual(results[0].reason_code, "catch_all_detected") |
|
|
| async def test_batch_preserves_input_order_and_handles_syntax_failures(self): |
| provider = FakeBatchProvider( |
| results={"ok@corp.test": SMTPResult(verified=True, smtp_code=250)}, |
| catch_all={"corp.test": False}, |
| ) |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable_clean()), |
| ): |
| results = await engine.validate_batch( |
| ["not-an-email", "ok@corp.test"], deep=True |
| ) |
|
|
| self.assertEqual(len(results), 2) |
| self.assertEqual(results[0].email, "not-an-email") |
| self.assertEqual(results[0].status, ValidationStatus.INVALID) |
| self.assertEqual(results[1].email, "ok@corp.test") |
| self.assertEqual(results[1].status, ValidationStatus.VALID) |
|
|
| async def test_batch_falls_back_to_single_when_provider_cannot_batch(self): |
| class SingleOnlyProvider: |
| def __init__(self): |
| self.calls = 0 |
|
|
| async def verify(self, email, mx_host): |
| self.calls += 1 |
| return SMTPResult(verified=True, smtp_code=250) |
|
|
| async def catch_all(self, domain, mx_host): |
| return (False, 0.9) |
|
|
| provider = SingleOnlyProvider() |
| engine = EmailValidationEngine(mailbox=provider, cache=NullCache()) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test", "b@corp.test"], deep=True) |
|
|
| self.assertEqual(len(results), 2) |
| self.assertEqual(provider.calls, 2) |
|
|
|
|
| class BulkRouteBatchTests(unittest.IsolatedAsyncioTestCase): |
| async def test_process_bulk_sync_uses_the_batched_path(self): |
| from app.api import routes |
|
|
| called = {} |
|
|
| async def fake_batch(emails, deep=True): |
| called["emails"] = list(emails) |
| called["deep"] = deep |
|
|
| class R: |
| def model_dump(self): |
| return {"email": "x"} |
|
|
| return [R() for _ in emails] |
|
|
| with patch.object(routes.engine, "validate_batch", new=AsyncMock(side_effect=fake_batch)): |
| out = await routes.process_bulk_sync(["a@b.com", "c@d.com"], deep=True) |
|
|
| self.assertEqual(called["emails"], ["a@b.com", "c@d.com"]) |
| self.assertTrue(called["deep"]) |
| self.assertEqual(len(out), 2) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
