File size: 4,921 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 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 | import unittest
from app.validator.smtp_client import SmtpConnection, RcptOutcome
class FakeSock:
"""Scripted TCP socket: returns queued byte chunks on recv, records sends."""
def __init__(self, chunks):
self.chunks = list(chunks)
self.sent: list[str] = []
self.closed = False
self.timeout = None
def settimeout(self, t):
self.timeout = t
def sendall(self, data: bytes):
self.sent.append(data.decode("utf-8", "replace"))
def recv(self, _n: int) -> bytes:
return self.chunks.pop(0) if self.chunks else b""
def close(self):
self.closed = True
def make_conn(chunks, **kw):
sock = FakeSock(chunks)
conn = SmtpConnection(
"mx.example.test",
timeout=5,
helo_domain="validator.test",
connect=lambda *a, **k: sock,
**kw,
)
return conn, sock
BANNER = b"220 mx.example.test ESMTP ready\r\n"
EHLO_OK = b"250-mx.example.test\r\n250-PIPELINING\r\n250 OK\r\n"
class SmtpConnectionTests(unittest.TestCase):
def test_handshake_reads_banner_and_sends_ehlo(self):
conn, sock = make_conn([BANNER, EHLO_OK, b"250 OK\r\n"])
conn.open()
self.assertTrue(conn.ready)
self.assertTrue(any(c.startswith("EHLO ") for c in sock.sent))
self.assertIn("validator.test", "".join(sock.sent))
def test_multiline_ehlo_is_parsed_as_single_response(self):
conn, _ = make_conn([BANNER, EHLO_OK])
conn.open()
self.assertEqual(conn.server_code, 250)
def test_accepting_rcpt_returns_250(self):
conn, sock = make_conn(
[BANNER, EHLO_OK, b"250 OK\r\n", b"250 2.1.5 Ok\r\n", b"250 OK\r\n"]
)
conn.open()
out = conn.probe("real@example.test")
self.assertEqual(out.code, 250)
self.assertTrue(out.accepted)
self.assertFalse(out.transport_error)
self.assertTrue(any(c.startswith("RCPT TO:") for c in sock.sent))
def test_rejecting_rcpt_returns_550(self):
conn, _ = make_conn(
[
BANNER,
EHLO_OK,
b"250 OK\r\n",
b"550 5.1.1 The email account that you tried to reach does not exist\r\n",
b"250 OK\r\n",
]
)
conn.open()
out = conn.probe("ghost@example.test")
self.assertEqual(out.code, 550)
self.assertFalse(out.accepted)
self.assertIn("does not exist", out.message)
def test_multiple_probes_reuse_one_connection(self):
# The whole point of the engine: one TCP session, many RCPT probes.
conn, sock = make_conn(
[
BANNER,
EHLO_OK,
b"250 OK\r\n", # MAIL FROM #1
b"250 OK\r\n", # RCPT #1
b"250 OK\r\n", # RSET
b"250 OK\r\n", # MAIL FROM #2
b"550 5.1.1 nope\r\n", # RCPT #2
b"250 OK\r\n", # RSET
b"221 Bye\r\n",
]
)
conn.open()
first = conn.probe("a@example.test")
second = conn.probe("b@example.test")
conn.quit()
self.assertTrue(first.accepted)
self.assertFalse(second.accepted)
# Two MAIL FROM commands, one EHLO, one connection.
sends = "".join(sock.sent)
self.assertEqual(sends.count("MAIL FROM:"), 2)
self.assertEqual(sends.count("EHLO "), 1)
self.assertEqual(sends.count("RSET"), 2)
self.assertTrue(sock.closed)
def test_transport_failure_is_reported_not_raised(self):
conn, _ = make_conn([BANNER, EHLO_OK])
conn.open()
conn._sock.chunks = [] # server dies mid-session
out = conn.probe("a@example.test")
self.assertTrue(out.transport_error)
self.assertIsNone(out.code)
self.assertTrue(out.retryable)
def test_unexpected_banner_is_not_ready(self):
conn, _ = make_conn([b"554 Service unavailable\r\n"])
conn.open()
self.assertFalse(conn.ready)
def test_greylist_code_marked_retryable(self):
conn, _ = make_conn(
[BANNER, EHLO_OK, b"250 OK\r\n", b"451 4.7.1 Greylisted, try later\r\n", b"250 OK\r\n"]
)
conn.open()
out = conn.probe("a@example.test")
self.assertEqual(out.code, 451)
self.assertTrue(out.retryable)
self.assertFalse(out.permanent_reject)
class RcptOutcomeTests(unittest.TestCase):
def test_flags_for_5xx(self):
o = RcptOutcome(code=550, message="550 no such user")
self.assertTrue(o.permanent_reject)
self.assertFalse(o.retryable)
def test_flags_for_accept(self):
o = RcptOutcome(code=250, message="250 OK")
self.assertTrue(o.accepted)
self.assertFalse(o.retryable)
self.assertFalse(o.permanent_reject)
if __name__ == "__main__":
unittest.main() |