If our application can send e-mail (and, according to Letts’ Law, it eventually will), we would want our integration tests to verify that the e-mail has indeed been sent, and that its contents are as expected. If we have written the integration tests in Python this will be pretty straightforward: Python library provides smtpd package with an implementation of SMTP server class, all we need to do is extend it to store the received messages and provide a way for the test code to verify that message has indeed been received. The Mock server class presented below runs in a separate thread and can be started and interrogated from existing test scripts.
'''
Provides a mock SMTP server implementation, MockSMTPServer.
Sample usage:
----
# create the server -- will start automatically
import smtpmock
mock_server = smtpmock.MockSMTPServer("localhost", 25025)
#send a test message
import smtplib
client = smtplib.SMTP("localhost", 25025)
fromaddr = "test.sender@mydomain.com"
toaddrs = ["test.recipient1@mydomain.com", "test.recipient2@mydomain.com"]
content = "test message content"
msg = "From: %s\r\nTo: %s\r\n\r\n%s" % (fromaddr, ", ".join(toaddrs), content)
client.sendmail(fromaddr, toaddrs, msg)
client.quit()
# verify that the message has been recieved
assert(mock_server.received_message_matching("From: .*\\nTo: .*\\n+.+tent"))
# reset the server to be ready for a new test
mock_server.reset()
assert(mock_server.received_messages_count() == 0)
----
'''
import asyncore
import re
import smtpd
import threading
class MockSMTPServer(smtpd.SMTPServer, threading.Thread):
'''
A mock SMTP server. Runs in a separate thread so can be started from
existing test code.
'''
def __init__(self, hostname, port):
threading.Thread.__init__(self)
smtpd.SMTPServer.__init__(self, (hostname, port), None)
self.daemon = True
self.received_messages = []
self.start()
def run(self):
asyncore.loop()
def process_message(self, peer, mailfrom, rcpttos, data):
self.received_messages.append(data)
def reset(self):
self.received_messages = []
# helper methods for assertions in test cases
def received_message_matching(self, template):
for message in self.received_messages:
if re.match(template, message): return True
return False
def received_messages_count(self):
return len(self.received_messages)