| 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): |
| |
| conn, sock = make_conn( |
| [ |
| BANNER, |
| EHLO_OK, |
| b"250 OK\r\n", |
| b"250 OK\r\n", |
| b"250 OK\r\n", |
| b"250 OK\r\n", |
| b"550 5.1.1 nope\r\n", |
| b"250 OK\r\n", |
| 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) |
| |
| 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 = [] |
| 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() |
