Setting up a fake SMTPD server for pytest with pytest-asyncio, aiosmtpd and aiosmtplib

test_smtpd.py

from collections import namedtuple
from functools import partial

import pytest
from aiosmtpd.handlers import Debugging as SMTPDDebuggingHandler
from aiosmtpd.smtp import SMTP as SMTPD
from aiosmtplib import SMTP

RecordedMessage = namedtuple(
    "RecordedMessage",
    "peer envelope_from envelope_recipients data",
)


class MessageHandler(SMTPDDebuggingHandler):

    messages = None

    def __init__(self, *args, **kwargs):
        self.messages = []
        super().__init__(*args, **kwargs)

    def handle_DATA(self, server, session, envelope):
        print(f"SMTP GOT MESSAGE: {envelope.content}")

        msg = RecordedMessage(
            session.peer, envelope.mail_from, envelope.rcpt_tos, envelope.content
        )
        self.messages.append(msg)

        return super().handle_DATA(server, session, envelope)


@pytest.fixture
def smtpd_port(unused_tcp_port):
    return unused_tcp_port


@pytest.fixture
async def smtpd(request, event_loop, smtpd_port):
    handler = MessageHandler()
    factory = partial(SMTPD, handler=handler)

    server = await event_loop.create_server(factory, port=smtpd_port)

    server.message_handler = handler

    yield server

    server.close()


@pytest.fixture
async def smtp_client(request, smtpd, smtpd_port):
    smtp_client = SMTP(port=smtpd_port)
    return smtp_client