| import unittest |
| from unittest.mock import patch |
|
|
| from app.validator.egress import EgressReport, check_port25_egress |
|
|
|
|
| class FakeSock: |
| def __init__(self, banner=b"220 mx.google.com ESMTP ready\r\n", fail=None): |
| self._banner = banner |
| self._fail = fail |
| self.sent: list[str] = [] |
| self.closed = False |
| self._read = False |
|
|
| def settimeout(self, t): |
| pass |
|
|
| def sendall(self, data): |
| self.sent.append(data.decode("utf-8", "replace")) |
|
|
| def recv(self, _n): |
| if self._fail: |
| raise self._fail |
| if self._read: |
| return b"250 OK\r\n" |
| self._read = True |
| return self._banner |
|
|
| def close(self): |
| self.closed = True |
|
|
|
|
| def _connector(sock): |
| def _connect(host, port, timeout): |
| return sock |
| return _connect |
|
|
|
|
| class EgressPreflightTests(unittest.TestCase): |
| """Tells you, before you trust a host, whether outbound port 25 works — |
| the single prerequisite for real mailbox verification.""" |
|
|
| def test_reachable_host_reports_success(self): |
| sock = FakeSock() |
| report = check_port25_egress(connect=_connector(sock)) |
| self.assertTrue(report.reachable) |
| self.assertTrue(report.ok) |
| self.assertIn("220", report.banner) |
|
|
| def test_timeout_is_reported_as_blocked_with_actionable_advice(self): |
| import socket as _s |
|
|
| def _connect(host, port, timeout): |
| raise _s.timeout("timed out") |
|
|
| report = check_port25_egress(connect=_connect) |
| self.assertFalse(report.reachable) |
| self.assertFalse(report.ok) |
| |
| self.assertIn("block", report.diagnosis.lower()) |
| self.assertIn("port 25", report.advice.lower()) |
|
|
| def test_refused_reports_blocked(self): |
| def _connect(host, port, timeout): |
| raise ConnectionRefusedError("refused") |
|
|
| report = check_port25_egress(connect=_connect) |
| self.assertFalse(report.ok) |
| self.assertIn("25", report.advice) |
|
|
| def test_unreachable_network_reports_blocked(self): |
| def _connect(host, port, timeout): |
| raise OSError(101, "Network is unreachable") |
|
|
| report = check_port25_egress(connect=_connect) |
| self.assertFalse(report.ok) |
| self.assertTrue(report.diagnosis) |
|
|
| def test_helo_accepted_is_recorded(self): |
| sock = FakeSock() |
| report = check_port25_egress(connect=_connector(sock), helo_domain="validator.test") |
| self.assertTrue(report.helo_accepted) |
| self.assertTrue(any("EHLO" in s for s in sock.sent)) |
|
|
| def test_socket_closed_after_probe(self): |
| sock = FakeSock() |
| check_port25_egress(connect=_connector(sock)) |
| self.assertTrue(sock.closed, "preflight must not leak the connection") |
|
|
| def test_report_serialises_for_ops_output(self): |
| report = check_port25_egress(connect=_connector(FakeSock())) |
| data = report.as_dict() |
| self.assertIn("reachable", data) |
| self.assertIn("advice", data) |
|
|
| def test_default_target_is_a_reliable_mx(self): |
| report = check_port25_egress(connect=_connector(FakeSock())) |
| self.assertIn("google", report.target.lower()) |
|
|
|
|
| class EgressAdviceTests(unittest.TestCase): |
| def test_advice_names_the_hosts_that_block_port_25(self): |
| def _connect(host, port, timeout): |
| raise TimeoutError("timed out") |
|
|
| report = check_port25_egress(connect=_connect) |
| |
| self.assertTrue( |
| any( |
| name in report.advice.lower() |
| for name in ("hugging face", "aws", "paas") |
| ), |
| f"advice should name blocking platforms, got: {report.advice}", |
| ) |
| self.assertIn("vps", report.advice.lower()) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
