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 ) # Two unique domains -> two DNS resolutions, not three. 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")], ) # The per-address path must NOT be used when batching is available. 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()