pytest fixtures for SMTP and subprocess

test_integration_smtp.py

import asyncio
from email.message import EmailMessage
from email.parser import BytesParser as EmailParser
from email.policy import default as default_policy
from queue import Queue, Empty
from smtplib import SMTP as SMTPClient
from typing import NamedTuple, List, Tuple

from aiosmtpd.controller import Controller as SMTPController
import pytest


class SMTPMessage(NamedTuple):
    remote_host: Tuple[str, int]
    sender: str
    receivers: List[str]
    message: str


def queue_consumer(queue: Queue):
    while True:
        try:
            yield queue.get(timeout=5)
        except Empty:
            break


@pytest.fixture
def smtp_messages():
    message_q: Queue = Queue()

    class SMTPHandler:
        async def handle_DATA(self, _server, session, envelope):
            peer = session.peer
            mail_from = envelope.mail_from
            rcpt_tos = envelope.rcpt_tos
            data = envelope.content  # type: bytes
            msg = SMTPMessage(peer, mail_from, rcpt_tos, data)
            message_q.put(msg)
            return '250 OK'

    controller = SMTPController(SMTPHandler(), hostname='127.0.0.1', port=10025)
    controller.start()
    try:
        yield queue_consumer(message_q)
    finally:
        controller.stop()


def test_message_is_received(smtp_messages):
    message = EmailMessage()
    message['Subject'] = 'Test Email'
    message['From'] = '[email protected]'
    message['To'] = ['[email protected]', '[email protected]']
    message.set_content('Hello!')
    
    client = SMTPClient(host='127.0.0.1', port=10025)
    client.send_message(message)
    client.quit()

    for received_smtp in smtp_messages:
        received_email = EmailParser(policy=default_policy).parsebytes(received_smtp.message)
        assert received_smtp.sender == message['From']
        assert received_email['from'] == message['From']
        assert received_smtp.receivers == message['To']
        assert received_email['to'] == ', '.join(message['To'])
        assert received_smtp.remote_host[0] == '127.0.0.1'
        assert received_email['subject'] == message['Subject']
        assert received_email.get_content().strip() == message['body']
        break

test_integration_subprocess.py

from threading import Thread
import os
from queue import Queue, Empty
from subprocess import Popen, PIPE
from typing import BinaryIO, List, Tuple


def queue_consumer(queue: Queue):
    while True:
        try:
            yield queue.get(timeout=5)
        except Empty:
            break


def read_output(out_stream: BinaryIO, out_queue: Queue) -> None:
    for line in iter(out_stream.readline, b''):
        out_queue.put(line.decode('utf-8'))  # Change this to your program out encoding
    out_stream.close()

            
def make_program(command: List[str], extra_env: dict) -> Tuple[Popen, Thread, Queue]:
    env = os.environ.copy()
    env.update(extra_env)
    queue: Queue = Queue()
    sub_p = Popen(command, stdout=PIPE, env=env)
    reader_t = Thread(target=read_output, args=(sub_p.stdout, queue))
    reader_t.daemon = True
    queue_gen = queue_consumer(queue)
    return sub_p, reader_t, queue_gen


@pytest.fixture
def program_out():
    extra_env = {
        'MY_ENV_VAR': '42'
    }
    command = ['program', 'arg1', 'arg2']
    sub_p, reader_t, queue_gen = make_program(command, extra_env)
    reader_t.start()
    try:
        yield queue_gen
    finally:
        sub_p.terminate()
        reader_t.join(5)

def test_program_output_line_by_line(program_out):
    for line in program_out:
        # Do your tests and `break` when done
    else:
        assert False, 'Reached end of output'