FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | import unittest | |
| from unittest.mock import patch | |
| from app.config import settings | |
| from app.models import SMTPResult | |
| from app.validator.mailbox_provider import ( | |
| AutonomousSmtpProvider, | |
| LocalSmtpProvider, | |
| RemoteSmtpWorkerProvider, | |
| resolve_provider, | |
| ) | |
| class AutonomousProviderTests(unittest.IsolatedAsyncioTestCase): | |
| """The engine must stand alone: direct SMTP is the default, and no | |
| third-party key may be required for it to function.""" | |
| async def test_verify_uses_batch_engine(self): | |
| captured = {} | |
| class FakeBatch: | |
| def verify_batch(self, pairs): | |
| captured["pairs"] = list(pairs) | |
| return {e: SMTPResult(verified=True, smtp_code=250) for e, _ in pairs} | |
| p = AutonomousSmtpProvider(batch=FakeBatch()) | |
| out = await p.verify("a@corp.test", "mx.corp.test") | |
| self.assertTrue(out.verified) | |
| self.assertEqual(captured["pairs"], [("a@corp.test", "mx.corp.test")]) | |
| async def test_catch_all_probe_delegates_to_batch(self): | |
| class FakeBatch: | |
| def verify_batch(self, pairs): | |
| # The gibberish probe is rejected -> not catch-all. | |
| return {e: SMTPResult(verified=False, smtp_code=550) for e, _ in pairs} | |
| p = AutonomousSmtpProvider(batch=FakeBatch()) | |
| is_ca, confidence = await p.catch_all("corp.test", "mx.corp.test") | |
| self.assertIs(is_ca, False) | |
| self.assertGreater(confidence, 0.5) | |
| async def test_catch_all_unknown_on_transport_failure(self): | |
| class FakeBatch: | |
| def verify_batch(self, pairs): | |
| return {e: SMTPResult(verified=None, error="transport_error") for e, _ in pairs} | |
| p = AutonomousSmtpProvider(batch=FakeBatch()) | |
| is_ca, confidence = await p.catch_all("corp.test", "mx.corp.test") | |
| self.assertIsNone(is_ca) | |
| self.assertEqual(confidence, 0.0) | |
| async def test_catch_all_true_when_gibberish_accepted(self): | |
| class FakeBatch: | |
| def verify_batch(self, pairs): | |
| return { | |
| e: SMTPResult(verified=None, error="catch_all: corp.test") | |
| for e, _ in pairs | |
| } | |
| p = AutonomousSmtpProvider(batch=FakeBatch()) | |
| is_ca, confidence = await p.catch_all("corp.test", "mx.corp.test") | |
| self.assertIs(is_ca, True) | |
| self.assertGreater(confidence, 0.5) | |
| class DefaultProviderTests(unittest.TestCase): | |
| def test_auto_without_worker_resolves_to_autonomous(self): | |
| with patch.object(settings, "mailbox_provider", "auto"), patch.object( | |
| settings, "smtp_worker_url", "" | |
| ): | |
| self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider) | |
| def test_local_forced_resolves_to_autonomous(self): | |
| with patch.object(settings, "mailbox_provider", "local"): | |
| self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider) | |
| def test_auto_with_worker_resolves_remote(self): | |
| with patch.object(settings, "mailbox_provider", "auto"), patch.object( | |
| settings, "smtp_worker_url", "https://worker.example.com" | |
| ): | |
| self.assertIsInstance(resolve_provider(), RemoteSmtpWorkerProvider) | |
| def test_legacy_local_provider_still_importable(self): | |
| self.assertTrue(callable(LocalSmtpProvider)) | |
| if __name__ == "__main__": | |
| unittest.main() |