| import unittest |
| from unittest.mock import AsyncMock, patch |
|
|
| from app.models import ( |
| CatchAllResult, |
| DNSResult, |
| DisposableResult, |
| SMTPResult, |
| SyntaxResult, |
| ) |
| from app.api.routes import process_bulk_sync |
| from app.validator.engine import EmailValidationEngine |
|
|
|
|
| def _syntax(**kw): |
| base = dict( |
| passed=True, |
| normalized_email="person@example.test", |
| domain="example.test", |
| local_part="person", |
| ) |
| base.update(kw) |
| return SyntaxResult(**base) |
|
|
|
|
| def _dns(**kw): |
| base = dict( |
| domain_exists=True, |
| has_mx_records=True, |
| mx_records=[{"host": "mx.example.test", "priority": 10}], |
| ) |
| base.update(kw) |
| return DNSResult(**base) |
|
|
|
|
| class BulkDepthTests(unittest.IsolatedAsyncioTestCase): |
| """RC1: the bulk path hard-coded deep=False, skipping catch-all + SMTP, so |
| every bulk row collapsed to Unknown/risky/insufficient_depth in the UI. |
| Bulk must run the same deep pipeline as single verification.""" |
|
|
| async def test_process_bulk_sync_runs_deep_layers(self): |
| """Bulk must run deep, via the batched path (one call for the chunk).""" |
| captured = {} |
|
|
| class FakeResult: |
| def __init__(self, email): |
| self.email = email |
|
|
| def model_dump(self): |
| return {"email": self.email} |
|
|
| async def capture_batch(emails, deep=True): |
| captured["emails"] = list(emails) |
| captured["deep"] = deep |
| return [FakeResult(e) for e in emails] |
|
|
| with patch( |
| "app.api.routes.engine.validate_batch", |
| new=AsyncMock(side_effect=capture_batch), |
| ): |
| out = await process_bulk_sync(["a@example.com", "b@example.com"]) |
|
|
| self.assertEqual(captured["emails"], ["a@example.com", "b@example.com"]) |
| self.assertTrue(captured["deep"], "bulk must run deep") |
| self.assertEqual(len(out), 2) |
|
|
| async def test_engine_deep_runs_catch_all_and_smtp(self): |
| engine = EmailValidationEngine() |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=DisposableResult( |
| is_disposable=False, is_free_provider=False, is_role_based=False |
| )), |
| patch( |
| "app.validator.engine.resolve_provider", |
| return_value=None, |
| ), |
| ): |
| pass |
|
|
| async def test_injected_mailbox_provider_is_used_for_deep(self): |
| """The engine must call the injected provider's catch_all + verify.""" |
| calls = {"verify": 0, "catch_all": 0} |
|
|
| class FakeProvider: |
| async def verify(self, email, mx_host): |
| calls["verify"] += 1 |
| return SMTPResult(verified=True, smtp_code=250) |
|
|
| async def catch_all(self, domain, mx_host): |
| calls["catch_all"] += 1 |
| return (False, 0.9) |
|
|
| engine = EmailValidationEngine(mailbox=FakeProvider()) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=DisposableResult( |
| is_disposable=False, is_free_provider=False, is_role_based=False |
| )), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(calls["verify"], 1) |
| self.assertEqual(calls["catch_all"], 1) |
| self.assertTrue(result.smtp.verified) |
| self.assertIn("smtp", result.checks_run) |
| self.assertIn("catch_all", result.checks_run) |
|
|
|
|
| class BulkRequestDeepFieldTests(unittest.TestCase): |
| def test_bulk_request_accepts_deep_flag(self): |
| from app.models import BulkEmailRequest |
|
|
| req = BulkEmailRequest(emails=["a@example.com"], deep=True) |
| self.assertTrue(req.deep) |
|
|
| def test_bulk_request_defaults_deep_true(self): |
| |
| from app.models import BulkEmailRequest |
|
|
| req = BulkEmailRequest(emails=["a@example.com"]) |
| self.assertTrue(req.deep) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
