SMTP server pytest fixture

smtpd_fixture.py

# The SMTP server runs in a separate thread and stores sent messages in a list
#
# Usage:
#
# def test_fn(smtpd):
#    host = 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 == ['[email protected]']
#

from collections import namedtuple
from smtpd import SMTPServer
from threading import Lock, Thread
import asyncore
import time

import pytest


RecordedMessage = namedtuple(
    'RecordedMessage',
    '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:
            self._items.append(item)

    def copy(self):
        with self._lock:
            return self._items[:]


class SMTPServerThread(Thread):
    def __init__(self, messages):
        super().__init__()
        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)
                _messages.add(msg)

        self.smtp = _SMTPServer(('127.0.0.1', 0), None)
        self.host_port = self.smtp.getsockname()
        asyncore.loop(timeout=0.1)

    def close(self):
        self.smtp.close()


class SMTPServerFixture:
    def __init__(self):
        self._messages = ThreadSafeList()
        self._thread = SMTPServerThread(self._messages)
        self._thread.start()

    @property
    def host_port(self):
        '''SMTP server's listening address as a (host, port) tuple'''
        while self._thread.host_port is None:
            time.sleep(0.1)
        return self._thread.host_port

    @property
    def host(self):
        return self.host_port[0]

    @property
    def port(self):
        return self.host_port[1]

    @property
    def messages(self):
        '''A list of RecordedMessage objects'''
        return self._messages.copy()

    def close(self):
        self._thread.close()
        self._thread.join(10)
        if self._thread.is_alive():
            raise RuntimeError('smtp server thread did not stop in 10 sec')


@pytest.fixture
def smtpd(request):
    fixture = SMTPServerFixture()
    request.addfinalizer(fixture.close)
    return fixture

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.

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

You're right, ThreadSafeList is probably unnecessary.

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()