File size: 10,183 Bytes
f5afa05 1a43824 03c7b2c 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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | import unittest
from app.validator.mx_batch import CatchAllCache, MxBatchVerifier
from app.validator.smtp_client import RcptOutcome
class FakeConnection:
"""Records probes and answers from a scripted map.
Keyed by lowercased address; a "*" entry is the default answer, which is how
a catch-all domain behaves (accepts everything).
"""
registry: dict[str, dict] = {}
opened: list[str] = []
closed: list[int] = []
def __init__(self, host, **kwargs):
self.host = host
self.answers = FakeConnection.registry.get(host, {})
self.ready = True
self.opened.append(host)
def open(self):
self.ready = True
return True
def probe(self, email):
# Catch-all probe uses a gibberish local part -> "*" rule applies.
answer = self.answers.get(email.lower(), self.answers.get("*"))
if answer is None:
return RcptOutcome(code=550, message="550 5.1.1 no such user")
if isinstance(answer, RcptOutcome):
return answer
return RcptOutcome(code=answer, message=f"{answer} response")
def quit(self):
FakeConnection.closed.append(1)
def close(self):
pass
class MxBatchVerifierTests(unittest.TestCase):
def setUp(self):
FakeConnection.registry = {}
FakeConnection.opened = []
FakeConnection.closed = []
def verifier(self, **kw):
return MxBatchVerifier(
connection_factory=FakeConnection,
max_connections_per_host=kw.pop("max_connections_per_host", 2),
**kw,
)
def test_groups_by_mx_and_opens_one_session_per_host(self):
# google host: two addresses; corp host: one address
FakeConnection.registry = {
"gmail-smtp-in.l.google.com": {"a@gmail.com": 250, "b@gmail.com": 550, "*": 550},
"mx.corp.test": {"c@corp.test": 250, "*": 550},
}
v = self.verifier()
out = v.verify_batch(
[
("a@gmail.com", "gmail-smtp-in.l.google.com"),
("b@gmail.com", "gmail-smtp-in.l.google.com"),
("c@corp.test", "mx.corp.test"),
]
)
self.assertEqual(out["a@gmail.com"].verified, True)
self.assertEqual(out["b@gmail.com"].verified, False)
self.assertEqual(out["c@corp.test"].verified, True)
# One connection opened per MX host (not per address).
self.assertEqual(sorted(FakeConnection.opened), ["gmail-smtp-in.l.google.com", "mx.corp.test"])
def test_catch_all_domain_does_not_certify_mailbox(self):
# Microsoft 365 behaviour: accepts everything at RCPT.
FakeConnection.registry = {
"x.mail.protection.outlook.com": {"*": 250},
}
v = self.verifier()
out = v.verify_batch([("anyone@contoso.com", "x.mail.protection.outlook.com")])
result = out["anyone@contoso.com"]
# Accepted by an accept-all host, so NOT verified.
self.assertIsNone(result.verified)
self.assertIn("catch_all", (result.error or "").lower())
def test_catch_all_probe_is_cached_per_domain(self):
FakeConnection.registry = {"mx.corp.test": {"*": 250}}
v = self.verifier()
v.verify_batch([("a@corp.test", "mx.corp.test")])
probes_after_first = len(FakeConnection.opened)
# Second address in the same domain must reuse the cached catch-all flag.
v.verify_batch([("b@corp.test", "mx.corp.test")])
self.assertEqual(len(FakeConnection.opened), probes_after_first)
def test_transport_error_marks_unknown_not_invalid(self):
class DeadConn(FakeConnection):
def open(self):
self.ready = False
return False
FakeConnection.registry = {"mx.dead.test": {"*": 250}}
v = MxBatchVerifier(connection_factory=DeadConn)
out = v.verify_batch([("a@dead.test", "mx.dead.test")])
r = out["a@dead.test"]
self.assertIsNone(r.verified)
self.assertIsNotNone(r.error)
def test_greylisted_address_is_reported_retryable(self):
FakeConnection.registry = {
"mx.corp.test": {"a@corp.test": RcptOutcome(code=451, message="451 greylisted"), "*": 550}
}
v = self.verifier()
out = v.verify_batch([("a@corp.test", "mx.corp.test")])
self.assertIsNone(out["a@corp.test"].verified)
def test_empty_batch_returns_empty(self):
v = self.verifier()
self.assertEqual(v.verify_batch([]), {})
class CatchAllCacheTests(unittest.TestCase):
def test_stores_and_returns_within_ttl(self):
c = CatchAllCache(ttl_seconds=60)
c.put("corp.test", True)
self.assertIs(c.get("corp.test"), True)
def test_expires_after_ttl(self):
c = CatchAllCache(ttl_seconds=0)
c.put("corp.test", True)
self.assertIsNone(c.get("corp.test"))
def test_unknown_is_not_cached_as_false(self):
c = CatchAllCache(ttl_seconds=60)
c.put("corp.test", None)
# None (probe failed) must not be cached as a decision.
self.assertIsNone(c.get("corp.test"))
def test_distinguishes_false_from_missing(self):
c = CatchAllCache(ttl_seconds=60)
c.put("strict.test", False)
self.assertIs(c.get("strict.test"), False)
self.assertIsNone(c.get("never-seen.test"))
class EgressCircuitBreakerTests(unittest.TestCase):
"""When outbound port 25 is blocked (HF Spaces, most PaaS) every connect
burns the full timeout. A bulk job must detect that once and fail fast
instead of spending timeout × N — 10k rows at 10s/row is ~27 hours."""
def setUp(self):
FakeConnection.registry = {}
FakeConnection.opened = []
def test_circuit_opens_after_threshold_and_stops_connecting(self):
class DeadConn(FakeConnection):
def open(self):
self.ready = False
return False
FakeConnection.registry = {"mx.a.test": {"*": 250}}
v = MxBatchVerifier(connection_factory=DeadConn, fail_threshold=2, cooldown_seconds=60)
v.verify_batch([("a@a.test", "mx.a.test")])
v.verify_batch([("b@b.test", "mx.b.test")])
self.assertTrue(v.circuit_open)
before = len(FakeConnection.opened)
out = v.verify_batch([("c@c.test", "mx.c.test")])
# No new connection attempted while the circuit is open.
self.assertEqual(len(FakeConnection.opened), before)
self.assertIsNone(out["c@c.test"].verified)
self.assertIn("unreachable", (out["c@c.test"].error or "").lower())
def test_successful_connection_resets_failure_count(self):
class FlakyConn(FakeConnection):
fail_next = True
def open(self):
if FlakyConn.fail_next:
self.ready = False
return False
self.ready = True
return True
FakeConnection.registry = {"mx.a.test": {"*": 250}}
v = MxBatchVerifier(connection_factory=FlakyConn, fail_threshold=2, cooldown_seconds=60)
v.verify_batch([("a@a.test", "mx.a.test")])
FlakyConn.fail_next = False
v.verify_batch([("b@b.test", "mx.b.test")])
self.assertFalse(v.circuit_open)
def test_default_threshold_is_conservative(self):
v = MxBatchVerifier(connection_factory=FakeConnection)
self.assertGreaterEqual(v.fail_threshold, 2)
class ParallelDomainTests(unittest.IsolatedAsyncioTestCase):
"""Domains must be probed CONCURRENTLY (with a cap).
Sequential probing means wall-clock = sum of every domain's latency; on a
blocked host that is timeout x domains. Commercial validators get bulk
throughput from parallelism across domains, not speed per domain.
"""
async def test_domains_are_probed_in_parallel(self):
import asyncio
import time
active = {"now": 0, "peak": 0}
class SlowConn(FakeConnection):
def open(self):
active["now"] += 1
active["peak"] = max(active["peak"], active["now"])
time.sleep(0.25)
active["now"] -= 1
self.ready = True
return True
FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(8)}
v = MxBatchVerifier(
connection_factory=SlowConn, bulk_concurrency=8, max_connections_per_host=1
)
pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(8)]
t0 = time.time()
out = await v.verify_batch_async(pairs)
elapsed = time.time() - t0
self.assertEqual(len(out), 8)
# 8 x 0.25s sequential = 2.0s; parallel should be well under 1s.
self.assertLess(elapsed, 1.2, f"domains ran sequentially ({elapsed:.2f}s)")
self.assertGreater(active["peak"], 2, "no real concurrency observed")
async def test_concurrency_cap_is_respected(self):
import time
active = {"now": 0, "peak": 0}
class SlowConn(FakeConnection):
def open(self):
active["now"] += 1
active["peak"] = max(active["peak"], active["now"])
time.sleep(0.15)
active["now"] -= 1
self.ready = True
return True
FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(10)}
v = MxBatchVerifier(
connection_factory=SlowConn, bulk_concurrency=3, max_connections_per_host=1
)
pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(10)]
out = await v.verify_batch_async(pairs)
self.assertEqual(len(out), 10)
self.assertLessEqual(active["peak"], 3, "exceeded the concurrency cap")
async def test_sync_verify_batch_still_works(self):
FakeConnection.registry = {"mx.corp.test": {"a@corp.test": 250, "*": 550}}
v = MxBatchVerifier(connection_factory=FakeConnection)
out = v.verify_batch([("a@corp.test", "mx.corp.test")])
self.assertTrue(out["a@corp.test"].verified)
if __name__ == "__main__":
unittest.main() |