| import unittest |
| from unittest.mock import AsyncMock, patch |
|
|
| from app.models import DNSResult, DisposableResult, SMTPResult, ValidationStatus |
| from app.validator.engine import EmailValidationEngine |
|
|
|
|
| def _dns(host="mx.example.test"): |
| return DNSResult( |
| domain_exists=True, |
| has_mx_records=True, |
| mx_records=[{"host": host, "priority": 10}], |
| ) |
|
|
|
|
| def _clean(): |
| return DisposableResult( |
| is_disposable=False, is_free_provider=False, is_role_based=False |
| ) |
|
|
|
|
| class RecordingCache: |
| """In-memory stand-in for ValidationCache that records interactions.""" |
|
|
| def __init__(self): |
| self.dns: dict[str, DNSResult] = {} |
| self.verdicts: dict[tuple[str, bool], tuple[SMTPResult, object]] = {} |
| self.dns_puts: list[str] = [] |
| self.verdict_puts: list[str] = [] |
|
|
| async def get_dns(self, domain): |
| return self.dns.get(domain.lower()) |
|
|
| async def put_dns(self, domain, result): |
| if result.error_code in {"timeout", "resolver_error"}: |
| return |
| self.dns_puts.append(domain) |
| self.dns[domain.lower()] = result |
|
|
| async def get_verdict(self, email, deep): |
| return self.verdicts.get((email.lower(), deep)) |
|
|
| async def put_verdict(self, email, deep, smtp, catch_all): |
| if smtp.verified is None and catch_all is not True: |
| return |
| self.verdict_puts.append(email) |
| self.verdicts[(email.lower(), deep)] = (smtp, catch_all) |
|
|
|
|
| class CountingProvider: |
| def __init__(self, smtp_map=None): |
| self._smtp = smtp_map or {} |
| self.batch_calls = 0 |
| self.pairs_seen: list[tuple[str, str]] = [] |
|
|
| async def verify_many(self, pairs): |
| self.batch_calls += 1 |
| self.pairs_seen.extend(pairs) |
| return ( |
| {e: self._smtp.get(e, SMTPResult(verified=True, smtp_code=250)) for e, _ in pairs}, |
| {}, |
| ) |
|
|
| async def verify(self, email, mx_host): |
| return self._smtp.get(email, SMTPResult(verified=True, smtp_code=250)) |
|
|
| async def catch_all(self, domain, mx_host): |
| return (False, 0.9) |
|
|
|
|
| class EngineCacheTests(unittest.IsolatedAsyncioTestCase): |
| """The batch pipeline must consult and populate the cache.""" |
|
|
| async def test_dns_cache_hit_skips_resolution(self): |
| cache = RecordingCache() |
| cache.dns["corp.test"] = _dns() |
| engine = EmailValidationEngine(mailbox=CountingProvider(), cache=cache) |
|
|
| with ( |
| patch( |
| "app.validator.engine.check_dns", new=AsyncMock(return_value=_dns()) |
| ) as dns_mock, |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertEqual(dns_mock.await_count, 0, "cache hit should skip DNS") |
| self.assertEqual(results[0].status, ValidationStatus.VALID) |
|
|
| async def test_dns_miss_is_populated_for_next_run(self): |
| cache = RecordingCache() |
| engine = EmailValidationEngine(mailbox=CountingProvider(), cache=cache) |
|
|
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertIn("corp.test", cache.dns_puts) |
|
|
| async def test_dns_result_is_reused_across_addresses_in_one_run(self): |
| cache = RecordingCache() |
| engine = EmailValidationEngine(mailbox=CountingProvider(), cache=cache) |
|
|
| with ( |
| patch( |
| "app.validator.engine.check_dns", new=AsyncMock(return_value=_dns()) |
| ) as dns_mock, |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| await engine.validate_batch( |
| ["a@corp.test", "b@corp.test", "c@corp.test"], deep=True |
| ) |
|
|
| self.assertEqual(dns_mock.await_count, 1) |
|
|
| async def test_cached_verdict_skips_the_mailbox_probe(self): |
| cache = RecordingCache() |
| cache.dns["corp.test"] = _dns() |
| cache.verdicts[("a@corp.test", True)] = ( |
| SMTPResult(verified=True, smtp_code=250), |
| False, |
| ) |
| provider = CountingProvider() |
| engine = EmailValidationEngine(mailbox=provider, cache=cache) |
|
|
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertEqual(provider.batch_calls, 0, "cached verdict should skip the probe") |
| self.assertEqual(results[0].status, ValidationStatus.VALID) |
| self.assertTrue(results[0].smtp.verified) |
|
|
| async def test_only_uncached_addresses_are_probed(self): |
| cache = RecordingCache() |
| cache.dns["corp.test"] = _dns() |
| cache.verdicts[("a@corp.test", True)] = ( |
| SMTPResult(verified=True, smtp_code=250), |
| False, |
| ) |
| provider = CountingProvider() |
| engine = EmailValidationEngine(mailbox=provider, cache=cache) |
|
|
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| await engine.validate_batch(["a@corp.test", "b@corp.test"], deep=True) |
|
|
| self.assertEqual(provider.pairs_seen, [("b@corp.test", "mx.example.test")]) |
|
|
| async def test_definitive_verdict_is_cached(self): |
| cache = RecordingCache() |
| engine = EmailValidationEngine( |
| mailbox=CountingProvider( |
| {"a@corp.test": SMTPResult(verified=False, smtp_code=550, |
| smtp_message="550 no such user")} |
| ), |
| cache=cache, |
| ) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertIn("a@corp.test", cache.verdict_puts) |
|
|
| async def test_inconclusive_verdict_is_not_cached(self): |
| cache = RecordingCache() |
| engine = EmailValidationEngine( |
| mailbox=CountingProvider( |
| {"a@corp.test": SMTPResult(verified=None, error="transport_error")} |
| ), |
| cache=cache, |
| ) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertNotIn("a@corp.test", cache.verdict_puts) |
|
|
| async def test_cached_row_still_reports_all_seven_layers(self): |
| cache = RecordingCache() |
| cache.dns["corp.test"] = _dns() |
| cache.verdicts[("a@corp.test", True)] = ( |
| SMTPResult(verified=True, smtp_code=250), |
| False, |
| ) |
| engine = EmailValidationEngine(mailbox=CountingProvider(), cache=cache) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test"], deep=True) |
|
|
| self.assertEqual( |
| results[0].checks_run, |
| ["syntax", "dns", "mx", "disposable", "catch_all", "smtp", "scoring"], |
| ) |
|
|
| async def test_fast_mode_does_not_read_deep_verdicts(self): |
| cache = RecordingCache() |
| cache.dns["corp.test"] = _dns() |
| cache.verdicts[("a@corp.test", True)] = ( |
| SMTPResult(verified=True, smtp_code=250), |
| False, |
| ) |
| engine = EmailValidationEngine(mailbox=CountingProvider(), cache=cache) |
| with ( |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_clean()), |
| ): |
| results = await engine.validate_batch(["a@corp.test"], deep=False) |
|
|
| |
| |
| self.assertIsNone(results[0].smtp) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
