File size: 3,388 Bytes
f5afa05 de745bf f5afa05 de745bf f5afa05 de745bf f5afa05 de745bf f5afa05 de745bf f5afa05 de745bf f5afa05 | 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 | 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() |