122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
|
import imaplib
|
||
|
|
import smtplib
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from proxy_server import IMAPBackend, SMTPProxyHandler, Settings, env_bool
|
||
|
|
|
||
|
|
|
||
|
|
class DummyIMAP:
|
||
|
|
def __init__(self, host, port):
|
||
|
|
self.host = host
|
||
|
|
self.port = port
|
||
|
|
self.logged_in = False
|
||
|
|
self.selected = None
|
||
|
|
|
||
|
|
def login(self, username, password):
|
||
|
|
self.logged_in = True
|
||
|
|
|
||
|
|
def select(self, mailbox):
|
||
|
|
self.selected = mailbox
|
||
|
|
|
||
|
|
def uid(self, command, *args):
|
||
|
|
if command == "search":
|
||
|
|
return "OK", [b"1 2 3"]
|
||
|
|
if command == "fetch" and args[1] == "(RFC822.SIZE)":
|
||
|
|
return "OK", [(b"1 (RFC822.SIZE 1024)", b"")]
|
||
|
|
if command == "fetch" and args[1] == "(RFC822)":
|
||
|
|
return "OK", [b"1 (RFC822 {10}", b"Hello", b" World", b")"]
|
||
|
|
return "NO", []
|
||
|
|
|
||
|
|
def logout(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
def expunge(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def test_env_bool_interprets_truthy_values(monkeypatch):
|
||
|
|
monkeypatch.setenv("FEATURE_ENABLED", "true")
|
||
|
|
assert env_bool("FEATURE_ENABLED") is True
|
||
|
|
monkeypatch.setenv("FEATURE_ENABLED", "1")
|
||
|
|
assert env_bool("FEATURE_ENABLED") is True
|
||
|
|
monkeypatch.setenv("FEATURE_ENABLED", "off")
|
||
|
|
assert env_bool("FEATURE_ENABLED") is False
|
||
|
|
monkeypatch.delenv("FEATURE_ENABLED", raising=False)
|
||
|
|
assert env_bool("FEATURE_ENABLED", default=True) is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_settings_validate_requires_backend_hosts():
|
||
|
|
original_imap = Settings.BACKEND_IMAP_HOST
|
||
|
|
original_smtp = Settings.BACKEND_SMTP_HOST
|
||
|
|
Settings.BACKEND_IMAP_HOST = None
|
||
|
|
Settings.BACKEND_SMTP_HOST = None
|
||
|
|
with pytest.raises(RuntimeError, match="Missing required environment variables"):
|
||
|
|
Settings.validate()
|
||
|
|
Settings.BACKEND_IMAP_HOST = original_imap
|
||
|
|
Settings.BACKEND_SMTP_HOST = original_smtp
|
||
|
|
|
||
|
|
|
||
|
|
def test_settings_validate_succeeds_with_backends():
|
||
|
|
original_imap = Settings.BACKEND_IMAP_HOST
|
||
|
|
original_smtp = Settings.BACKEND_SMTP_HOST
|
||
|
|
Settings.BACKEND_IMAP_HOST = "imap.example.com"
|
||
|
|
Settings.BACKEND_SMTP_HOST = "smtp.example.com"
|
||
|
|
Settings.validate()
|
||
|
|
Settings.BACKEND_IMAP_HOST = original_imap
|
||
|
|
Settings.BACKEND_SMTP_HOST = original_smtp
|
||
|
|
|
||
|
|
|
||
|
|
def test_imap_backend_can_login_and_fetch(monkeypatch):
|
||
|
|
monkeypatch.setattr(imaplib, "IMAP4_SSL", lambda host, port: DummyIMAP(host, port))
|
||
|
|
backend = IMAPBackend("user", "pass")
|
||
|
|
backend.login()
|
||
|
|
assert backend.connection.logged_in
|
||
|
|
assert backend.connection.selected == "INBOX"
|
||
|
|
assert backend.list_uids() == [b"1", b"2", b"3"]
|
||
|
|
assert backend.fetch_message_size(b"1") == 1024
|
||
|
|
assert b"Hello" in backend.fetch_message(b"1")
|
||
|
|
|
||
|
|
|
||
|
|
def test_smtp_proxy_handler_forwards_message_over_ssl(monkeypatch):
|
||
|
|
captured = {}
|
||
|
|
|
||
|
|
class DummySMTP:
|
||
|
|
def __init__(self, host, port, timeout=0):
|
||
|
|
captured["instance"] = self
|
||
|
|
self.host = host
|
||
|
|
self.port = port
|
||
|
|
self.timeout = timeout
|
||
|
|
self.logged_in = False
|
||
|
|
self.sent = None
|
||
|
|
|
||
|
|
def login(self, user, password):
|
||
|
|
self.logged_in = True
|
||
|
|
|
||
|
|
def sendmail(self, sender, recipients, message):
|
||
|
|
self.sent = (sender, recipients, message)
|
||
|
|
|
||
|
|
def quit(self):
|
||
|
|
captured["quit"] = True
|
||
|
|
|
||
|
|
monkeypatch.setattr(smtplib, "SMTP_SSL", DummySMTP)
|
||
|
|
previous_ssl = Settings.BACKEND_SMTP_USE_SSL
|
||
|
|
previous_user = Settings.BACKEND_SMTP_USER
|
||
|
|
previous_pass = Settings.BACKEND_SMTP_PASS
|
||
|
|
Settings.BACKEND_SMTP_USE_SSL = True
|
||
|
|
Settings.BACKEND_SMTP_USER = "smtp-user"
|
||
|
|
Settings.BACKEND_SMTP_PASS = "smtp-pass"
|
||
|
|
|
||
|
|
handler = SMTPProxyHandler()
|
||
|
|
handler.send_message("from@example.com", ["to@example.com"], b"Subject: Test\r\n\r\nBody\r\n")
|
||
|
|
|
||
|
|
assert captured["instance"].host == Settings.BACKEND_SMTP_HOST
|
||
|
|
assert captured["instance"].logged_in is True
|
||
|
|
assert captured["instance"].sent[0] == "from@example.com"
|
||
|
|
assert captured["instance"].sent[1] == ["to@example.com"]
|
||
|
|
assert b"Subject: Test" in captured["instance"].sent[2]
|
||
|
|
|
||
|
|
Settings.BACKEND_SMTP_USE_SSL = previous_ssl
|
||
|
|
Settings.BACKEND_SMTP_USER = previous_user
|
||
|
|
Settings.BACKEND_SMTP_PASS = previous_pass
|