File size: 3,374 Bytes
9649ce6 d82b7e3 9649ce6 | 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 | import unittest
from unittest.mock import AsyncMock, MagicMock, patch
from app.models import DNSResult, ValidationStatus
from app.validator.engine import EmailValidationEngine
def _dns_with_mx(host):
return DNSResult(
domain_exists=True,
has_mx_records=True,
mx_records=[{"host": host, "priority": 10}],
)
def _clean():
from app.models import DisposableResult
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 FakeProvider:
async def verify_many(self, pairs):
return ({e: SMTPResult(verified=True, smtp_code=250) for e, _ in pairs}, {})
async def verify(self, email, mx_host):
return SMTPResult(verified=True, smtp_code=250)
async def catch_all(self, domain, mx_host):
return (False, 0.9)
def _fake_model(answer):
m = MagicMock()
m.predict = MagicMock(return_value=[("DISPOSABLE", 0.99)])
return m
class TrustedMxGateTests(unittest.IsolatedAsyncioTestCase):
"""A domain running on a trusted enterprise mail platform (Google Workspace,
M365, ...) is almost never disposable — disposable domains live on cheap
hosts. The ML model learned 'unfamiliar domain string = disposable' from an
imbalanced dataset, so for trusted-MX domains the model's opinion must be
ignored: the deterministic MX intelligence is the better signal."""
def setUp(self):
self.patchers = [
patch("app.validator.engine.get_model_async",
new=AsyncMock(return_value=_fake_model("DISPOSABLE"))),
]
for p in self.patchers:
p.start()
def tearDown(self):
for p in self.patchers:
p.stop()
def _engine(self):
return EmailValidationEngine(mailbox=FakeProvider(), cache=NullCache())
async def test_google_workspace_mx_skips_ml_single(self):
engine = self._engine()
with patch("app.validator.engine.check_dns",
new=AsyncMock(return_value=_dns_with_mx("aspmx.l.google.com"))):
result = await engine.validate("info@thefreshfactory.co", deep=True)
self.assertIsNone(result.disposable.ml_label)
self.assertNotIn(
"ML_DISPOSABLE_SUSPECT",
{k for k in result.scoring.breakdown},
)
async def test_unknown_mx_still_gets_ml_signal(self):
engine = self._engine()
with patch("app.validator.engine.check_dns",
new=AsyncMock(return_value=_dns_with_mx("mx.sketchy-host.test"))):
result = await engine.validate("a@sketchy.test", deep=True)
self.assertEqual(result.disposable.ml_label, "DISPOSABLE")
async def test_m365_mx_skips_ml_bulk(self):
engine = self._engine()
with patch(
"app.validator.engine.check_dns",
new=AsyncMock(return_value=_dns_with_mx("x.mail.protection.outlook.com")),
):
results = await engine.validate_batch(["a@contoso.com"], deep=True)
self.assertIsNone(results[0].disposable.ml_label) |