pygrader/test/server.py

   1 # Copyright
   2
   3 import asyncore as _asyncore
   4 import email as _email
   5 import smtpd as _smptd
   6 import socket as _socket
   7
   8 from .. import LOG as _LOG
   9
  10
  11 class SMTPChannel (_smptd.SMTPChannel):
  12     def close(self):
  13         super(SMTPChannel, self).close()
  14         _LOG.debug('close {}'.format(self))
  15         self.smtp_server.channel_closed()
  16
  17
  18 class SMTPServer (_smptd.SMTPServer):
  19     """An SMTP server for testing pygrader.
  20
  21     >>> from asyncore import loop
  22     >>> from smtplib import SMTP
  23     >>> from pgp_mime.email import encodedMIMEText
  24     >>> from pygrader.test.client import MessageSender
  25
  26     >>> def process(peer, mailfrom, rcpttos, data):
  27     ...     print('peer:     {}'.format(peer))
  28     ...     print('mailfrom: {}'.format(mailfrom))
  29     ...     print('rcpttos:  {}'.format(rcpttos))
  30     ...     print('message:')
  31     ...     print(data)
  32     >>> server = SMTPServer(
  33     ...     ('localhost', 1025), None, process=process, count=3)
  34
  35     >>> message = encodedMIMEText('Ping')
  36     >>> message['From'] = 'a@example.com'
  37     >>> message['To'] = 'b@example.com, c@example.com'
  38     >>> message['Cc'] = 'd@example.com'
  39     >>> messages = [message, message, message]
  40     >>> ms = MessageSender(address=('localhost', 1025), messages=messages)
  41     >>> loop()  # doctest: +REPORT_UDIFF, +ELLIPSIS
  42     peer:     ('127.0.0.1', ...)
  43     mailfrom: a@example.com
  44     rcpttos:  ['b@example.com', 'c@example.com', 'd@example.com']
  45     message:
  46     Content-Type: text/plain; charset="us-ascii"
  47     MIME-Version: 1.0
  48     Content-Transfer-Encoding: 7bit
  49     Content-Disposition: inline
  50     From: a@example.com
  51     To: b@example.com, c@example.com
  52     Cc: d@example.com
  53     <BLANKLINE>
  54     Ping
  55     peer:     ('127.0.0.1', ...)
  56     mailfrom: a@example.com
  57     rcpttos:  ['b@example.com', 'c@example.com', 'd@example.com']
  58     message:
  59     Content-Type: text/plain; charset="us-ascii"
  60     MIME-Version: 1.0
  61     Content-Transfer-Encoding: 7bit
  62     Content-Disposition: inline
  63     From: a@example.com
  64     To: b@example.com, c@example.com
  65     Cc: d@example.com
  66     <BLANKLINE>
  67     Ping
  68     peer:     ('127.0.0.1', ...)
  69     mailfrom: a@example.com
  70     rcpttos:  ['b@example.com', 'c@example.com', 'd@example.com']
  71     message:
  72     Content-Type: text/plain; charset="us-ascii"
  73     MIME-Version: 1.0
  74     Content-Transfer-Encoding: 7bit
  75     Content-Disposition: inline
  76     From: a@example.com
  77     To: b@example.com, c@example.com
  78     Cc: d@example.com
  79     <BLANKLINE>
  80     Ping
  81     """
  82     channel_class = SMTPChannel
  83
  84     def __init__(self, *args, **kwargs):
  85         self.count = kwargs.pop('count', None)
  86         self.process = kwargs.pop('process', None)
  87         self.channels_open = 0
  88         super(SMTPServer, self).__init__(*args, **kwargs)
  89
  90     def log_info(self, message, type='info'):
  91         # TODO: type -> severity
  92         _LOG.info(message)
  93
  94     def handle_accepted(self, conn, addr):
  95         if self.count <= 0:
  96             conn.close()
  97             return
  98         super(SMTPServer, self).handle_accepted(conn, addr)
  99         self.channels_open += 1
 100
 101     def channel_closed(self):
 102         self.channels_open -= 1
 103         if self.channels_open == 0 and self.count <= 0:
 104             _LOG.debug('close {}'.format(self))
 105             self.close()
 106
 107     def process_message(self, peer, mailfrom, rcpttos, data):
 108         if self.count is not None:
 109             self.count -= 1
 110             _LOG.debug('Count: {}'.format(self.count))
 111         _LOG.debug('receiving message from: {}'.format(peer))
 112         _LOG.debug('message addressed from: {}'.format(mailfrom))
 113         _LOG.debug('message addressed to  : {}'.format(rcpttos))
 114         _LOG.debug('message length        : {}'.format(len(data)))
 115         if self.process:
 116             self.process(peer, mailfrom, rcpttos, data)
 117         return