Como no he encontrado algo conveniente en Python 3 para mis necesidades (correo parte de trenzado no se está ejecutando en Python 3), hice una pequeña maqueta con asyncio que se puede mejorar si se desea:
Definí un ImapProtocol que se extiende asyncio.Protocol. A continuación, poner en marcha un servidor de esta manera:
factory = loop.create_server(lambda: ImapProtocol(mailbox_map), 'localhost', 1143)
server = loop.run_until_complete(factory)
El mailbox_map es un mapa del mapa: correo electrónico -> mapa de buzones -> conjunto de mensajes. Entonces todos los mensajes/buzones están en la memoria.
Cada vez que un cliente se conecta, se crea una nueva instancia de ImapProtocol. A continuación, se ejecuta el ImapProtocol y respuestas para cada cliente, la implementación de la capacidad/login/FETCH/selección/búsqueda/tienda:
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...
Luego, en mis pruebas, iniciar el servidor durante la instalación con:
self.loop = asyncio.get_event_loop()
self.server = self.loop.run_until_complete(self.loop.create_server(create_imap_protocol, 'localhost', 12345))
Cuando quiero simular un nuevo mensaje:
imap_receive(Mail(to='[email protected]', mail_from='[email protected]', subject='hello'))
y detenerlo en el desmontaje:
self.server.close()
asyncio.wait_for(self.server.wait_closed(), 1)
cf https://github.com/bamthomas/aioimaplib/blob/master/aioimaplib/tests/imapserver.py
EDITAR: Tenía una parada con errores del servidor, Reescribí con asyncio.Protocolo y modificar la respuesta para reflejar los cambios
El enlace anterior está muerto, es probable que sea nuevo https://docs.python.org/2/library/smtpd.html – sdaau