Fake SMTP server for testing purposes

fake_mail.py

# -*- coding: utf8 -*-
import base64
import os
import tempfile
from smtpd import DebuggingServer

from json import loads, dumps


class TestSmtpServer(DebuggingServer):
    """ Почтовый сервер, заменяющий реальную отправку почты ее выводом
    и сохранением в фальшивом почтовом ящике """

    def process_message(self, peer, mailfrom, rcpttos, data):
        DebuggingServer.process_message(self, peer, mailfrom, rcpttos, data)
        headers, msg_meta, body, endboundary = data.split('\n\n')
        decoded_body = base64.decodestring(body)
        self.append_message_data_to_file(dict(
            peer=peer, mailfrom=mailfrom, rcpttos=rcpttos,
            data=data, decoded_body=decoded_body,
        ))

    def append_message_data_to_file(self, message_data):
        data = FakeMailBox.read_data() or []
        data.append(message_data)
        FakeMailBox.save_data(data)


class FakeMailBox(object):
    """ Фальшивый почтовый ящик, сохраняющий и читающий данные в формате JSON
    во временный файл / из временного файла """

    FILENAME = 'test_mail.json'

    @classmethod
    def read_data(cls):
        with cls.get_mail_temp_file('r') as f:
            contents = f.read()
        try:
            return loads(contents)
        except ValueError:
            return []

    @classmethod
    def save_data(cls, data):
        with cls.get_mail_temp_file('wb+') as f:
            f.write(dumps(data))
        try:
            os.chmod(cls.get_mail_temp_file_name(), 00777)
        except OSError:
            pass

    @classmethod
    def wipe(cls):
        cls.save_data([])

    @classmethod
    def get_mail_temp_file(cls, mode):
        return open(cls.get_mail_temp_file_name(), mode)

    @classmethod
    def get_mail_temp_file_name(cls):
        return os.path.join(tempfile.gettempdir(), cls.FILENAME)


if __name__ == '__main__':
    import asyncore
    localaddr = remoteaddr = ('localhost', 25)
    server = TestSmtpServer(localaddr, remoteaddr)
    try:
        asyncore.loop()
    except KeyboardInterrupt:
        pass