As I didn't find something convenient in python 3 for my needs (mail part of twisted is not running in python 3), I did a small mock with asyncio that you can improve if you'd like :
I defined an ImapProtocol which extends asyncio.Protocol. Then launch a server like this :
factory = loop.create_server(lambda: ImapProtocol(mailbox_map), 'localhost', 1143)
server = loop.run_until_complete(factory)
The mailbox_map is a map of map : email -> map of mailboxes -> set of messages. So all the messages/mailboxes are in memory.
Each time a client connects, a new instance of ImapProtocol is created.
Then, the ImapProtocol executes and answers for each client, implementing capability/login/fetch/select/search/store :
class ImapHandler(object):
def __init__(self, mailbox_map):
self.mailbox_map = mailbox_map
self.user_login = None
# ...
def connection_made(self, transport):
self.transport = transport
transport.write('* OK IMAP4rev1 MockIMAP Server ready\r\n'.encode())
def data_received(self, data):
command_array = data.decode().rstrip().split()
tag = command_array[0]
self.by_uid = False
self.exec_command(tag, command_array[1:])
def connection_lost(self, error):
if error:
log.error(error)
else:
log.debug('closing')
self.transport.close()
super().connection_lost(error)
def exec_command(self, tag, command_array):
command = command_array[0].lower()
if not hasattr(self, command):
return self.error(tag, 'Command "%s" not implemented' % command)
getattr(self, command)(tag, *command_array[1:])
def capability(self, tag, *args):
# code for it...
def login(self, tag, *args):
# code for it...
Then in my tests, I start the server during setup with :
self.loop = asyncio.get_event_loop()
self.server = self.loop.run_until_complete(self.loop.create_server(create_imap_protocol, 'localhost', 12345))
When I want to simulate a new message :
imap_receive(Mail(to='[email protected]', mail_from='[email protected]', subject='hello'))
And stop it at teardown :
self.server.close()
asyncio.wait_for(self.server.wait_closed(), 1)
cf https://github.com/bamthomas/aioimaplib/blob/master/aioimaplib/tests/imapserver.py
EDIT: I had a buggy stop of the server, I rewrote it with asyncio.Protocol and modify the answer to reflect the changes
