| 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): |
| |
| 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): |
| |
| 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) |
| |
| self.assertEqual(sorted(FakeConnection.opened), ["gmail-smtp-in.l.google.com", "mx.corp.test"]) |
|
|
| def test_catch_all_domain_does_not_certify_mailbox(self): |
| |
| 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"] |
| |
| 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) |
| |
| 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) |
| |
| 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")]) |
| |
| 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) |
| |
| 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() |
