File size: 4,462 Bytes
fd7800b 45aeb2f fd7800b 45aeb2f fd7800b 45aeb2f fd7800b 45aeb2f fd7800b 45aeb2f fd7800b 45aeb2f fd7800b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | 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 # covered by mailbox-injected tests below
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):
# Product promise: bulk runs the full 7-layer pipeline (1 credit/email).
from app.models import BulkEmailRequest
req = BulkEmailRequest(emails=["a@example.com"])
self.assertTrue(req.deep)
if __name__ == "__main__":
unittest.main()
|