| import unittest |
| from unittest.mock import AsyncMock, patch |
|
|
| from app.models import DNSResult, DisposableResult, SMTPResult, ValidationStatus |
| from app.validator.engine import EmailValidationEngine |
| from app.validator.retry_queue import MemoryRetryStore, RetryQueue |
|
|
|
|
| def _dns(host="mx.corp.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 NullCache: |
| 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 ScriptedProvider: |
| def __init__(self, mapping): |
| self.mapping = mapping |
|
|
| async def verify_many(self, pairs): |
| return ( |
| {e: self.mapping.get(e, SMTPResult(verified=True, smtp_code=250)) for e, _ in pairs}, |
| {}, |
| ) |
|
|
| async def verify(self, email, mx_host): |
| return self.mapping.get(email, SMTPResult(verified=True, smtp_code=250)) |
|
|
| async def catch_all(self, domain, mx_host): |
| return (False, 0.9) |
|
|
|
|
| class EngineRetryIntegrationTests(unittest.IsolatedAsyncioTestCase): |
| """4xx rows must be re-queued (bounded), definitive rows must not.""" |
|
|
| async def test_greylisted_row_is_enqueued(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=600, now=lambda: clock["t"]) |
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider( |
| {"a@corp.test": SMTPResult(verified=None, smtp_code=451, |
| smtp_message="451 greylisted, try later")} |
| ), |
| cache=NullCache(), |
| retry_queue=queue, |
| ) |
| 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(await queue.size(), 1) |
| entry = (await queue.due(limit=1)) if clock["t"] > 600 else None |
| |
| self.assertEqual(results[0].status, ValidationStatus.RISKY) |
|
|
| async def test_verified_row_is_not_enqueued(self): |
| queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0) |
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider( |
| {"a@corp.test": SMTPResult(verified=True, smtp_code=250)} |
| ), |
| cache=NullCache(), |
| retry_queue=queue, |
| ) |
| 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.assertEqual(await queue.size(), 0) |
|
|
| async def test_rejected_row_is_not_enqueued(self): |
| queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0) |
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider( |
| {"a@corp.test": SMTPResult(verified=False, smtp_code=550, |
| smtp_message="550 no such user")} |
| ), |
| cache=NullCache(), |
| retry_queue=queue, |
| ) |
| 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.assertEqual(await queue.size(), 0) |
|
|
| async def test_transport_failure_is_not_enqueued(self): |
| |
| queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0) |
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider( |
| {"a@corp.test": SMTPResult(verified=None, error="transport_error: timed out")} |
| ), |
| cache=NullCache(), |
| retry_queue=queue, |
| ) |
| 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.assertEqual(await queue.size(), 0) |
|
|
| async def test_queue_failure_does_not_break_validation(self): |
| class BrokenQueue: |
| async def enqueue(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider( |
| {"a@corp.test": SMTPResult(verified=None, smtp_code=451)} |
| ), |
| cache=NullCache(), |
| retry_queue=BrokenQueue(), |
| ) |
| 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(len(results), 1) |
|
|
| async def test_engine_without_queue_still_works(self): |
| engine = EmailValidationEngine( |
| mailbox=ScriptedProvider({"a@corp.test": SMTPResult(verified=True, smtp_code=250)}), |
| cache=NullCache(), |
| ) |
| 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].status, ValidationStatus.VALID) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
