File size: 3,919 Bytes
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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | import unittest
from app.validator.mx_intel import (
MxPolicy,
classify_mx,
parse_rcpt_code,
is_greylist_code,
is_permanent_reject,
should_retry,
)
class ClassifyMxTests(unittest.TestCase):
"""Recognising the receiving provider from its MX host lets the engine
apply the right policy: Google rejects unknown mailboxes at RCPT (trust it),
Microsoft 365 accepts everything and bounces later (never trust a 250)."""
def test_google_mx_is_strict(self):
for host in (
"gmail-smtp-in.l.google.com",
"alt1.aspmx.l.google.com",
"aspmx.l.google.com",
):
p = classify_mx(host)
self.assertEqual(p.provider, "google", host)
self.assertTrue(p.trusts_rcpt, host)
self.assertFalse(p.accepts_all, host)
def test_microsoft365_is_accept_all(self):
for host in (
"example-com.mail.protection.outlook.com",
"mail.protection.outlook.com",
):
p = classify_mx(host)
self.assertEqual(p.provider, "microsoft365", host)
self.assertTrue(p.accepts_all, host)
self.assertFalse(p.trusts_rcpt, host)
def test_mimecast_and_proofpoint_are_accept_all_gateways(self):
for host in (
"us-smtp-inbound-1.mimecast.com",
"mx0a-001b2d01.pphosted.com",
):
p = classify_mx(host)
self.assertTrue(p.accepts_all, host)
self.assertFalse(p.trusts_rcpt, host)
def test_yahoo_is_strict(self):
p = classify_mx("mta5.am0.yahoodns.net")
self.assertEqual(p.provider, "yahoo")
self.assertTrue(p.trusts_rcpt)
def test_unknown_host_defaults_to_neutral(self):
p = classify_mx("mx.example-corp.test")
self.assertEqual(p.provider, "unknown")
self.assertFalse(p.accepts_all)
# Unknown providers are not assumed untrustworthy, but a 250 alone does
# not prove the mailbox exists — the catch-all probe decides that.
self.assertFalse(p.trusts_rcpt)
def test_case_and_trailing_dot_insensitive(self):
self.assertEqual(classify_mx("ASPMX.L.GOOGLE.COM.").provider, "google")
class RcptCodeTests(unittest.TestCase):
def test_250_is_accept(self):
self.assertEqual(parse_rcpt_code("250 2.1.5 Ok"), 250)
def test_550_is_reject(self):
self.assertEqual(parse_rcpt_code("550 5.1.1 The email account does not exist"), 550)
def test_non_numeric_returns_none(self):
self.assertIsNone(parse_rcpt_code("garbage"))
def test_4xx_is_greylist(self):
self.assertTrue(is_greylist_code(421))
self.assertTrue(is_greylist_code(450))
self.assertTrue(is_greylist_code(451))
self.assertFalse(is_greylist_code(250))
self.assertFalse(is_greylist_code(550))
def test_5xx_is_permanent(self):
self.assertTrue(is_permanent_reject(550))
self.assertTrue(is_permanent_reject(551))
self.assertFalse(is_permanent_reject(450))
self.assertFalse(is_permanent_reject(None))
def test_retry_policy_only_for_greylist(self):
self.assertTrue(should_retry(450))
self.assertTrue(should_retry(451))
self.assertFalse(should_retry(550))
self.assertFalse(should_retry(250))
self.assertFalse(should_retry(None))
def test_connection_error_retries(self):
# A transport failure is transient and worth one retry.
self.assertTrue(should_retry(None, transport_error=True))
class MxPolicyTests(unittest.TestCase):
def test_policy_is_immutable_and_descriptive(self):
p = MxPolicy(provider="google", trusts_rcpt=True, accepts_all=False)
self.assertEqual(p.provider, "google")
with self.assertRaises(Exception):
p.provider = "other" # type: ignore[misc]
if __name__ == "__main__":
unittest.main() |