| 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) |
