Skip to content

Instantly share code, notes, and snippets.

Last active June 30, 2021 10:26
Show Gist options
  • Save akheron/cf3863cdc424f08929e4cb7dc365ef23 to your computer and use it in GitHub Desktop.
Save akheron/cf3863cdc424f08929e4cb7dc365ef23 to your computer and use it in GitHub Desktop.
SMTP server pytest fixture
# The SMTP server runs in a separate thread and stores sent messages in a list
# Usage:
# def test_fn(smtpd):
# host =
# port = smtpd.port
# # Run code that send email using an smtp server at host:port
# # Assert that correct messages were received
# assert len(smtpd.messages) == 5
# last_message = smtpd.messages[-1]
# assert last_message.recipients == ['']
from collections import namedtuple
from smtpd import SMTPServer
from threading import Lock, Thread
import asyncore
import time
import pytest
RecordedMessage = namedtuple(
'peer envelope_from envelope_recipients data',
class ThreadSafeList:
def __init__(self, *args, **kwds):
self._items = []
self._lock = Lock()
def add(self, item):
with self._lock:
def copy(self):
with self._lock:
return self._items[:]
class SMTPServerThread(Thread):
def __init__(self, messages):
self.messages = messages
self.host_port = None
def run(self):
_messages = self.messages
class _SMTPServer(SMTPServer):
def process_message(self, peer, mailfrom, rcpttos, data):
msg = RecordedMessage(peer, mailfrom, rcpttos, data)
self.smtp = _SMTPServer(('', 0), None)
self.host_port = self.smtp.getsockname()
def close(self):
class SMTPServerFixture:
def __init__(self):
self._messages = ThreadSafeList()
self._thread = SMTPServerThread(self._messages)
def host_port(self):
'''SMTP server's listening address as a (host, port) tuple'''
while self._thread.host_port is None:
return self._thread.host_port
def host(self):
return self.host_port[0]
def port(self):
return self.host_port[1]
def messages(self):
'''A list of RecordedMessage objects'''
return self._messages.copy()
def close(self):
if self._thread.is_alive():
raise RuntimeError('smtp server thread did not stop in 10 sec')
def smtpd(request):
fixture = SMTPServerFixture()
return fixture
Copy link

dlordi commented Aug 7, 2019

Nice work! What is exactly the purpose of ThreadSafeList?

Copy link

akheron commented Aug 8, 2019

The SMTP server runs in a different thread than test code. The point of ThreadSafeList is to ensure that the list of received messages is not corrupted if it's being read (by test code) amd written (by the SMTP server) at the same time.

Copy link

dlordi commented Aug 8, 2019

If I'm not wrong, list are thread safe in CPython. I guess this class opens to other Python implementations

Copy link

akheron commented Aug 8, 2019

You're right, ThreadSafeList is probably unnecessary.

Copy link

tjcim commented Jan 10, 2020

Thanks for the gist. I had to make a couple of changes to get this going:

  • line 14 - from last_message.recipients to last_message.envelope_recipients
  • line 56 - Added two arguments mail_options and rcpt_options
  • line 61 - from self.smtp.getsockname() to self.smtp.socket.getsockname()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment