[tor-commits] [bridgedb/master] Remove moved unittests from bridgedb.test.test_email_server.
isis at torproject.org
isis at torproject.org
Fri Jun 6 23:39:14 UTC 2014
commit c509a0abb5df5e355f1acc8185fe285e33763dca
Author: Isis Lovecruft <isis at torproject.org>
Date: Mon Jun 2 21:11:51 2014 +0000
Remove moved unittests from bridgedb.test.test_email_server.
---
lib/bridgedb/test/test_email_server.py | 507 +++++++++-----------------------
1 file changed, 140 insertions(+), 367 deletions(-)
diff --git a/lib/bridgedb/test/test_email_server.py b/lib/bridgedb/test/test_email_server.py
index 198cb27..f79f603 100644
--- a/lib/bridgedb/test/test_email_server.py
+++ b/lib/bridgedb/test/test_email_server.py
@@ -13,351 +13,99 @@
from __future__ import print_function
-import io
-import copy
-import os
-import shutil
import socket
+import string
import types
-from bridgedb.Dist import EmailBasedDistributor
-from bridgedb.email import server
-from bridgedb.parse.addr import BadEmail
-from bridgedb.persistent import Conf
-from bridgedb.schedule import Unscheduled
-from bridgedb.test.test_HTTPServer import DummyBridge
-from bridgedb.test.util import fileCheckDecorator
-from bridgedb.test.util import TestCaseMixin
-
from twisted.python import log
from twisted.internet import defer
from twisted.internet import reactor
+from twisted.mail.smtp import IMessage
from twisted.mail.smtp import SMTPBadRcpt
from twisted.mail.smtp import SMTPBadSender
from twisted.mail.smtp import User
from twisted.mail.smtp import Address
+from twisted.mail.smtp import rfc822
from twisted.test import proto_helpers
from twisted.trial import unittest
+from zope.interface import implementedBy
+from bridgedb.Dist import EmailBasedDistributor
+from bridgedb.Dist import TooSoonEmail
+from bridgedb.email import server
+from bridgedb.parse.addr import BadEmail
+from bridgedb.schedule import Unscheduled
+from bridgedb.test.util import fileCheckDecorator
+from bridgedb.test.util import TestCaseMixin
+from bridgedb.test.email_helpers import _createConfig
+from bridgedb.test.email_helpers import _createMailServerContext
-class CheckDKIMTests(unittest.TestCase):
- """Tests for :func:`email.server.checkDKIM`."""
+class SMTPMessageTests(unittest.TestCase):
+ """Unittests for :class:`bridgedb.email.server.SMTPMessage`."""
def setUp(self):
- """Create fake email, distributor, and associated context data."""
- self.goodMessage = io.StringIO(unicode("""\
-From: user at gmail.com
-To: bridges at localhost
-X-DKIM-Authentication-Results: pass
-Subject: testing
-
-get bridges
-"""))
- self.badMessage = io.StringIO(unicode("""\
-From: user at gmail.com
-To: bridges at localhost
-Subject: testing
-
-get bridges
-"""))
self.config = _createConfig()
- self.domainRules = self.config.EMAIL_DOMAIN_RULES
-
- def test_checkDKIM_good(self):
- message = server.smtp.rfc822.Message(self.goodMessage)
- self.assertTrue(server.checkDKIM(message,
- self.domainRules.get("gmail.com")))
-
- def test_checkDKIM_bad(self):
- message = server.smtp.rfc822.Message(self.badMessage)
- result = server.checkDKIM(message, self.domainRules.get("gmail.com"))
- self.assertIs(result, False)
-
+ self.context = _createMailServerContext(self.config)
+ self.message = server.SMTPMessage(self.context,
+ canonicalFromSMTP='localhost')
+ self.line = string.ascii_lowercase
-class CreateResponseBodyTests(unittest.TestCase):
- """Tests for :func:`bridgedb.email.server.createResponseBody`."""
-
- def _moveGPGTestKeyfile(self):
- here = os.getcwd()
- topDir = here.rstrip('_trial_temp')
- self.gpgFile = os.path.join(topDir, 'gnupghome', 'TESTING.subkeys.sec')
- self.gpgMoved = os.path.join(here, 'TESTING.subkeys.sec')
- shutil.copy(self.gpgFile, self.gpgMoved)
+ def tearDown(self):
+ """Re-enable safelogging in between test runs."""
+ server.safelog.setSafeLogging(True)
- def setUp(self):
- """Create fake email, distributor, and associated context data."""
- self._moveGPGTestKeyfile()
- self.toAddress = "user at example.com"
- self.config = _createConfig()
- self.ctx = _createMailContext(self.config)
- self.distributor = self.ctx.distributor
-
- def _getIncomingLines(self, clientAddress="user at example.com"):
- """Generate the lines of an incoming email from **clientAddress**."""
- self.toAddress = server.smtp.Address(clientAddress)
- lines = [
- "From: %s" % clientAddress,
- "To: bridges at localhost",
- "Subject: testing",
- "",
- "get bridges",
- ]
- return lines
-
- def test_createResponseBody_getKey(self):
- """A request for 'get key' should receive our GPG key."""
- lines = self._getIncomingLines()
- lines[4] = "get key"
- ret = server.createResponseBody(lines, self.ctx, self.toAddress)
- self.assertSubstring('-----BEGIN PGP PUBLIC KEY BLOCK-----', ret)
-
- def test_createResponseBody_bridges_invalid(self):
- """An invalid request for 'transport obfs3' should get help text."""
- lines = self._getIncomingLines("testing at localhost")
- lines[4] = "transport obfs3"
- ret = server.createResponseBody(lines, self.ctx, self.toAddress)
- self.assertSubstring("COMMANDs", ret)
-
- def test_createResponseBody_bridges_obfs3(self):
- """A request for 'get transport obfs3' should receive a response."""
- lines = self._getIncomingLines("testing at localhost")
- lines[4] = "get transport obfs3"
- ret = server.createResponseBody(lines, self.ctx, self.toAddress)
- self.assertSubstring("Here are your bridges", ret)
- self.assertSubstring("obfs3", ret)
-
- def test_createResponseBody_bridges_obfsobfswebz(self):
- """We should only pay attention to the *last* in a crazy request."""
- lines = self._getIncomingLines("testing at localhost")
- lines[4] = "get unblocked webz"
- lines.append("get transport obfs2")
- lines.append("get transport obfs3")
- ret = server.createResponseBody(lines, self.ctx, self.toAddress)
- self.assertSubstring("Here are your bridges", ret)
- self.assertSubstring("obfs3", ret)
-
- def test_createResponseBody_bridges_obfsobfswebzipv6(self):
- """We should *still* only pay attention to the *last* request."""
- lines = self._getIncomingLines("testing at localhost")
- lines[4] = "transport obfs3"
- lines.append("get unblocked webz")
- lines.append("get ipv6")
- lines.append("get transport obfs2")
- ret = server.createResponseBody(lines, self.ctx, self.toAddress)
- self.assertSubstring("Here are your bridges", ret)
- self.assertSubstring("obfs2", ret)
-
-
-class MailResponseTests(unittest.TestCase):
- """Tests for ``generateResponse()`` and ``MailResponse``."""
+ def test_SMTPMessage_init(self):
+ """Our ``message`` attribute should be a ``SMTPMessage`` object, and
+ ``message.responder`` should be a
+ :class:`bridgedb.email.autoresponder.SMTPAutoresponder`.
+ """
+ self.assertIsInstance(self.message, server.SMTPMessage)
+ self.assertIsInstance(self.message.responder,
+ server.autoresponder.SMTPAutoresponder)
- def setUp(self):
- self.fromAddr = "bridges at torproject.org"
- self.clientAddr = "user at example.com"
- self.body = """\
-People think that time is strictly linear, but, in reality, it's actually just
-a ball of timey-wimey, wibbly-warbly... stuff."""
+ def test_SMTPMessage_IMessage_interface(self):
+ """``SMTPMessage`` should implement ``twisted.mail.smtp.IMessage``."""
+ self.assertTrue(IMessage.implementedBy(server.SMTPMessage))
- def tearDown(self):
- server.safelog.safe_logging = True
-
- def test_MailResponse_generateResponse(self):
- response = server.generateResponse(self.fromAddr, self.clientAddr,
- self.body)
- self.assertIsInstance(response, server.MailResponse)
-
- def test_MailResponse_generateResponse_noSafelog(self):
- server.safelog.safe_logging = False
- response = server.generateResponse(self.fromAddr, self.clientAddr,
- self.body)
- self.assertIsInstance(response, server.MailResponse)
-
- def test_MailResponse_generateResponse_mailfile(self):
- response = server.generateResponse(self.fromAddr, self.clientAddr,
- self.body)
- self.assertIsInstance(response.mailfile, (io.BytesIO, io.StringIO))
-
- def test_MailResponse_generateResponse_withInReplyTo(self):
- response = server.generateResponse(self.fromAddr, self.clientAddr,
- self.body, messageID="NSA")
- contents = str(response.readContents()).replace('\x00', '')
- self.assertIsInstance(response.mailfile, (io.BytesIO, io.StringIO))
- self.assertSubstring("In-Reply-To: NSA", contents)
-
- def test_MailResponse_generateResponse_readContents(self):
- response = server.generateResponse(self.fromAddr, self.clientAddr,
- self.body)
- contents = str(response.readContents()).replace('\x00', '')
- self.assertSubstring('timey-wimey, wibbly-warbly... stuff.', contents)
-
- def test_MailResponse_additionalHeaders(self):
- response = server.MailResponse()
- response.writeHeaders(self.fromAddr, self.clientAddr,
- subject="Re: echelon", inReplyTo="NSA",
- X_been_there="They were so 2004")
- contents = str(response.readContents()).replace('\x00', '')
- self.assertIsInstance(response.mailfile, (io.BytesIO, io.StringIO))
- self.assertSubstring("In-Reply-To: NSA", contents)
- self.assertSubstring("X-been-there: They were so 2004", contents)
-
- def test_MailResponse_close(self):
- """Calling MailResponse.close() should close the ``mailfile`` and set
- ``closed=True``.
+ def test_SMTPMessage_lineReceived_withSafelog(self):
+ """Test sending a line of text to ``SMTPMessage.lineReceived`` with
+ safelogging enabled.
"""
- response = server.MailResponse()
- self.assertEqual(response.closed, False)
- response.close()
- self.assertEqual(response.closed, True)
- self.assertRaises(ValueError, response.write, self.body)
-
- def test_MailResponse_write(self):
- """Calling MailResponse.write() should write to the mailfile."""
- response = server.MailResponse()
- response.write(self.body)
- contents = str(response.readContents()).replace('\x00', '')
- self.assertEqual(self.body.replace('\n', '\r\n') + '\r\n', contents)
-
- def test_MailResponse_writelines(self):
- """Calling MailResponse.writelines() with a list should write the
- concatenated contents of the list into the mailfile.
+ server.safelog.setSafeLogging(True)
+ self.message.lineReceived(self.line)
+ self.assertEqual(self.message.nBytes, 26)
+ self.assertTrue(self.line in self.message.lines)
+
+ def test_SMTPMessage_lineReceived_withoutSafelog(self):
+ """Test sending a line of text to ``SMTPMessage.lineReceived`` with
+ safelogging disnabled.
"""
- response = server.MailResponse()
- response.writelines(self.body.split('\n'))
- contents = str(response.readContents()).replace('\x00', '')
- self.assertEqual(self.body.replace('\n', '\r\n') + '\r\n', contents)
+ server.safelog.setSafeLogging(False)
+ for _ in range(3):
+ self.message.lineReceived(self.line)
+ self.assertEqual(self.message.nBytes, 3*26)
+ self.assertTrue(self.line in self.message.lines)
+ def test_SMTPMessage_eomReceived(self):
+ """Calling ``oemReceived`` should return a deferred."""
+ self.message.lineReceived(self.line)
+ self.assertIsInstance(self.message.eomReceived(),
+ defer.Deferred)
-class MailMessageTests(unittest.TestCase):
- """Unittests for :class:`bridgedb.email.server.MailMessage`."""
+ def test_SMTPMessage_getIncomingMessage(self):
+ """``getIncomingMessage`` should return a ``rfc822.Message``."""
+ self.message.lineReceived(self.line)
+ self.assertIsInstance(self.message.getIncomingMessage(),
+ rfc822.Message)
- def setUp(self):
- self.config = _createConfig()
- self.context = _createMailContext(self.config)
- self.message = server.MailMessage(self.context)
-
- def _getIncomingLines(self, clientAddress="user at example.com"):
- """Generate the lines of an incoming email from **clientAddress**."""
- lines = [
- "From: %s" % clientAddress,
- "To: bridges at localhost",
- "Subject: testing",
- "",
- "get bridges",
- ]
- self.message.lines = lines
-
- def test_MailMessage_getMailFrom_notbridgedb_at_yikezors_dot_net(self):
- """MailMessage.getMailFrom() for an incoming email sent to any email
- address other than the one we're listening for should return our
- configured address, not the one in the incoming email.
- """
- self._getIncomingLines()
- self.message.lines[1] = 'To: notbridgedb at yikezors.net'
- incoming = self.message.getIncomingMessage()
- recipient = str(self.message.getMailFrom(incoming))
- self.assertEqual(recipient, self.context.fromAddr)
-
- def test_MailMessage_getMailFrom_givemebridges_at_seriously(self):
- """MailMessage.getMailFrom() for an incoming email sent to any email
- address other than the one we're listening for should return our
- configured address, not the one in the incoming email.
- """
- self._getIncomingLines()
- self.message.lines[1] = 'To: givemebridges at serious.ly'
- incoming = self.message.getIncomingMessage()
- recipient = str(self.message.getMailFrom(incoming))
- self.assertEqual(recipient, self.context.fromAddr)
-
- def test_MailMessage_getMailFrom_bad_address(self):
- """MailMessage.getMailFrom() for an incoming email sent to a malformed
- email address should log an smtp.AddressError and then return our
- configured email address.
- """
- self._getIncomingLines()
- self.message.lines[1] = 'To: ><@><<<>>.foo'
- incoming = self.message.getIncomingMessage()
- recipient = str(self.message.getMailFrom(incoming))
- self.assertEqual(recipient, self.context.fromAddr)
-
- def test_MailMessage_reply_noFrom(self):
- """A received email without a "From:" or "Sender:" header shouldn't
- receive a response.
- """
- self._getIncomingLines()
- self.message.lines[0] = ""
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_badAddress(self):
- """Don't respond to RFC2822 malformed source addresses."""
- self._getIncomingLines("testing*.?\"@example.com")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_anotherBadAddress(self):
- """Don't respond to RFC2822 malformed source addresses."""
- self._getIncomingLines("Mallory <>>@example.com")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_invalidDomain(self):
- """Don't respond to RFC2822 malformed source addresses."""
- self._getIncomingLines("testing at exa#mple.com")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_anotherInvalidDomain(self):
- """Don't respond to RFC2822 malformed source addresses."""
- self._getIncomingLines("testing at exam+ple.com")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_DKIM_badDKIMheader(self):
- """An email with an 'X-DKIM-Authentication-Result:' header appended
- after the body should not receive a response.
- """
- self._getIncomingLines("testing at gmail.com")
- self.message.lines.append("X-DKIM-Authentication-Result: ")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_goodDKIMheader(self):
- """An email with a good DKIM header should be responded to."""
- self._getIncomingLines("testing at gmail.com")
- self.message.lines.insert(3, "X-DKIM-Authentication-Result: pass")
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
-
- def test_MailMessage_reply_transport_invalid(self):
- """An invalid request for 'transport obfs3' should get help text."""
- self.skip = True
- raise unittest.SkipTest("We need to fake the reactor for this one")
-
- self._getIncomingLines("testing at example.com")
- self.message.lines[4] = "transport obfs3"
- ret = self.message.reply()
- self.assertSubstring("COMMANDs", ret)
-
- def test_MailMessage_reply_transport_valid(self):
- """An valid request for 'get transport obfs3' should get obfs3."""
- self.skip = True
- raise unittest.SkipTest("We need to fake the reactor for this one")
-
- self._getIncomingLines("testing at example.com")
- self.message.lines[4] = "transport obfs3"
- ret = self.message.reply()
- self.assertIsInstance(ret, defer.Deferred)
- self.assertSubstring("obfs3", ret)
- return ret
-
-
-class MailDeliveryTests(unittest.TestCase):
- """Unittests for :class:`email.server.MailDelivery`."""
+
+class SMTPIncomingDeliveryTests(unittest.TestCase):
+ """Unittests for :class:`email.server.SMTPIncomingDelivery`."""
def setUp(self):
- """Set up our :class:`server.MailDelivery` instance, and reset the
+ """Set up our :class:`server.SMTPIncomingDelivery` instance, and reset the
following ``TestCase`` attributes to ``None``:
- ``helo``
- ``proto``
@@ -365,8 +113,8 @@ class MailDeliveryTests(unittest.TestCase):
- ``user``
"""
self.config = _createConfig()
- self.context = _createMailContext(self.config)
- self.delivery = server.MailDelivery()
+ self.context = _createMailServerContext(self.config)
+ self.delivery = server.SMTPIncomingDelivery()
self.helo = None
self.proto = None
@@ -417,39 +165,41 @@ class MailDeliveryTests(unittest.TestCase):
"""
self.helo = ('localhost', '127.0.0.1')
self.origin = server.smtp.Address('client at example.com')
- self.delivery.setBridgeDBContext(self.context)
+ self.delivery.setContext(self.context)
- def _setUpRCPTTO(self, username=None):
+ def _setUpRCPTTO(self, username=None, domain=None, ip=None):
"""Set up the parameters for emulating a connected client sending a
SMTP 'RCPT TO:' command to us.
The default is to emulate sending: ``RCPT TO: bridges at localhost``.
"""
name = username if username is not None else self.config.EMAIL_USERNAME
- self._createUser(name, 'localhost', '127.0.0.1')
- self.delivery.setBridgeDBContext(self.context)
-
- def test_MailDelivery_init(self):
- """After calling :meth:`server.MailDelivery.__init__`, we should have a
- :class:`server.MailDelivery` object instance.
+ host = domain if domain is not None else 'localhost'
+ addr = ip if ip is not None else '127.0.0.1'
+ self._createUser(name, host, ip)
+ self.delivery.setContext(self.context)
+
+ def test_SMTPIncomingDelivery_init(self):
+ """After calling :meth:`server.SMTPIncomingDelivery.__init__`, we should have a
+ :class:`server.SMTPIncomingDelivery` object instance.
"""
- self.assertIsInstance(self.delivery, server.MailDelivery)
+ self.assertIsInstance(self.delivery, server.SMTPIncomingDelivery)
- def test_MailDelivery_setBridgeDBContext(self):
- """Calling :meth:`server.MailDelivery.setBridgeDBContext` should set
- the :ivar:`MailDelivery.context` attribute.
+ def test_SMTPIncomingDelivery_setContext(self):
+ """Calling :meth:`server.SMTPIncomingDelivery.setContext` should set
+ the :ivar:`SMTPIncomingDelivery.context` attribute.
- The ``MailDelivery.context`` should be a :class:`server.MailContext`,
+ The ``SMTPIncomingDelivery.context`` should be a :class:`server.MailServerContext`,
and it should have relevant settings from the config file stored
within it.
"""
- self.delivery.setBridgeDBContext(self.context)
- self.assertIsInstance(self.delivery.context, server.MailContext)
+ self.delivery.setContext(self.context)
+ self.assertIsInstance(self.delivery.context, server.MailServerContext)
self.assertEqual(self.delivery.context.smtpFromAddr,
self.config.EMAIL_SMTP_FROM_ADDR)
- def test_MailDelivery_receivedHeader(self):
- """The email resulting from a MailDelivery, the latter received from
+ def test_SMTPIncomingDelivery_receivedHeader(self):
+ """The email resulting from a SMTPIncomingDelivery, the latter received from
``'client at example.com'`` should contain a header stating:
``'Received: from example.com'``.
"""
@@ -457,20 +207,20 @@ class MailDeliveryTests(unittest.TestCase):
hdr = self.delivery.receivedHeader(self.helo, self.origin, [self.user,])
self.assertSubstring("Received: from example.com", hdr)
- def test_MailDelivery_validateFrom(self):
- """A valid origin should be stored as ``MailDelivery.fromCanonical``."""
+ def test_SMTPIncomingDelivery_validateFrom(self):
+ """A valid origin should be stored as ``SMTPIncomingDelivery.fromCanonical``."""
self._setUpMAILFROM()
self.delivery.validateFrom(self.helo, self.origin)
- self.assertEqual(self.delivery.fromCanonical, 'example.com')
+ self.assertEqual(self.delivery.fromCanonicalSMTP, 'example.com')
- def test_MailDelivery_validateFrom_unsupportedDomain(self):
+ def test_SMTPIncomingDelivery_validateFrom_unsupportedDomain(self):
"""A domain not in our canon should raise a SMTPBadSender."""
self._setUpMAILFROM()
origin = server.smtp.Address('throwing.pickles at yo.mama')
self.assertRaises(SMTPBadSender,
self.delivery.validateFrom, self.helo, origin)
- def test_MailDelivery_validateFrom_origin_notAdressType(self):
+ def test_SMTPIncomingDelivery_validateFrom_origin_notAdressType(self):
"""A non ``twisted.mail.smtp.Address`` origin should raise an
AttributeError exception.
"""
@@ -478,28 +228,48 @@ class MailDeliveryTests(unittest.TestCase):
origin = 'throwing.pickles at yo.mama'
self.delivery.validateFrom(self.helo, origin)
- def test_MailDelivery_validateTo(self):
- """Should return a callable that results in a MailMessage."""
+ def test_SMTPIncomingDelivery_validateTo(self):
+ """Should return a callable that results in a SMTPMessage."""
self._setUpRCPTTO()
validated = self.delivery.validateTo(self.user)
self.assertIsInstance(validated, types.FunctionType)
- self.assertIsInstance(validated(), server.MailMessage)
+ self.assertIsInstance(validated(), server.SMTPMessage)
- def test_MailDelivery_validateTo_plusAddress(self):
- """Should return a callable that results in a MailMessage."""
+ def test_SMTPIncomingDelivery_validateTo_plusAddress(self):
+ """Should return a callable that results in a SMTPMessage."""
self._setUpRCPTTO('bridges+ar')
validated = self.delivery.validateTo(self.user)
self.assertIsInstance(validated, types.FunctionType)
- self.assertIsInstance(validated(), server.MailMessage)
+ self.assertIsInstance(validated(), server.SMTPMessage)
- def test_MailDelivery_validateTo_badUsername(self):
- """A :class:`server.MailDelivery` which sends a SMTP
+ def test_SMTPIncomingDelivery_validateTo_badUsername_plusAddress(self):
+ """'givemebridges+zh_cn at ...' should raise an SMTPBadRcpt exception."""
+ self._setUpRCPTTO('givemebridges+zh_cn')
+ self.assertRaises(SMTPBadRcpt, self.delivery.validateTo, self.user)
+
+ def test_SMTPIncomingDelivery_validateTo_badUsername(self):
+ """A :class:`server.SMTPIncomingDelivery` which sends a SMTP
``RCPT TO: yo.mama at localhost`` should raise a
``twisted.mail.smtp.SMTPBadRcpt`` exception.
"""
self._setUpRCPTTO('yo.mama')
self.assertRaises(SMTPBadRcpt, self.delivery.validateTo, self.user)
+ def test_SMTPIncomingDelivery_validateTo_notOurDomain(self):
+ """An SMTP ``RCPT TO: bridges at forealsi.es`` should raise an SMTPBadRcpt
+ exception.
+ """
+ self._setUpRCPTTO('bridges', 'forealsi.es')
+ self.assertRaises(SMTPBadRcpt, self.delivery.validateTo, self.user)
+
+ def test_SMTPIncomingDelivery_validateTo_subdomain(self):
+ """An SMTP ``RCPT TO: bridges at subdomain.localhost`` should be allowed.
+ """
+ self._setUpRCPTTO('bridges', 'subdomain.localhost')
+ validated = self.delivery.validateTo(self.user)
+ self.assertIsInstance(validated, types.FunctionType)
+ self.assertIsInstance(validated(), server.SMTPMessage)
+
class SMTPTestCaseMixin(TestCaseMixin):
"""Utility methods for use within any subclasses of
@@ -516,7 +286,7 @@ class SMTPTestCaseMixin(TestCaseMixin):
class ExampleSMTPTests(SMTPTestCaseMixin, unittest.TestCase):
def setUp(self):
- factory = twisted.mail.smtp.SMTPFactory()
+ factory = twisted.mail.smtp.SMTPIncomingServerFactory()
self.proto = self.factory.buildProtocol(('127.0.0.1', 0))
@@ -550,7 +320,7 @@ class SMTPTestCaseMixin(TestCaseMixin):
segment. Includes the SMTP DATA EOM command ('.') at the end. If no
keyword arguments are given, the defaults are fairly sane.
- Suitable for testing a :class:`bridgedb.email.server.MailFactory`.
+ Suitable for testing a :class:`bridgedb.email.server.SMTPIncomingServerFactory`.
:param str fromAddr: An email address for the 'From:' header.
:param str toAddr: An email address for the 'To:' header.
@@ -611,14 +381,17 @@ class SMTPTestCaseMixin(TestCaseMixin):
self.assertSubstring(expected, recv)
-class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
- """Unittests for :class:`bridgedb.email.server.MailFactory`."""
+class SMTPIncomingServerFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
+ """Unittests for :class:`bridgedb.email.server.SMTPIncomingServerFactory`."""
def setUp(self):
- """Set up a localhost MailFactory handler incoming SMTP connections."""
+ """Set up a localhost SMTPIncomingServerFactory handler incoming SMTP
+ connections.
+ """
config = _createConfig()
- context = _createMailContext(config)
- factory = server.MailFactory(context)
+ context = _createMailServerContext(config)
+ factory = server.SMTPIncomingServerFactory()
+ factory.setContext(context)
factory.protocol.timeout = None # Otherwise the reactor gets dirty
self.smtpFromAddr = context.smtpFromAddr # 'bridges at localhost'
@@ -629,40 +402,40 @@ class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
self.transport.protocol = self.proto
self.proto.makeConnection(self.transport)
- def test_MailFactory_HELO_localhost(self):
+ def test_SMTPIncomingServerFactory_HELO_localhost(self):
"""Send 'HELO localhost' to the server's transport."""
ip = self.transport.getPeer().host
self._test(['HELO localhost'],
"Hello %s, nice to meet you" % ip)
- def test_MailFactory_MAIL_FROM_testing_at_localhost(self):
+ def test_SMTPIncomingServerFactory_MAIL_FROM_testing_at_localhost(self):
"""Send 'MAIL FROM: human at localhost'."""
self._test(['HELO localhost',
'MAIL FROM: testing at localhost'],
"250 Sender address accepted")
- def test_MailFactory_MAIL_FROM_testing_at_gethostname(self):
+ def test_SMTPIncomingServerFactory_MAIL_FROM_testing_at_gethostname(self):
"""Send 'MAIL FROM: human at hostname' for the local hostname."""
hostname = socket.gethostname() or "computer"
self._test(['HELO localhost',
'MAIL FROM: testing@%s' % hostname],
"250 Sender address accepted")
- def test_MailFactory_MAIL_FROM_testing_at_ipaddress(self):
+ def test_SMTPIncomingServerFactory_MAIL_FROM_testing_at_ipaddress(self):
"""Send 'MAIL FROM: human at ipaddr' for the loopback IP address."""
- hostname = socket.gethostbyname(socket.gethostname()) or "127.0.0.1"
+ hostname = 'localhost'
self._test(['HELO localhost',
'MAIL FROM: testing@%s' % hostname],
"250 Sender address accepted")
- def test_MailFactory_RCPT_TO_config_EMAIL_SMTP_FROM_ADDR(self):
+ def test_SMTPIncomingServerFactory_RCPT_TO_context_smtpFromAddr(self):
"""Send 'RCPT TO:' with the context.smtpFromAddr."""
self._test(['HELO localhost',
'MAIL FROM: testing at localhost',
'RCPT TO: %s' % self.smtpFromAddr],
"250 Recipient address accepted")
- def test_MailFactory_DATA_blank(self):
+ def test_SMTPIncomingServerFactory_DATA_blank(self):
"""A DATA command with nothing after it should receive::
'354 Continue'
in response.
@@ -673,7 +446,7 @@ class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
"DATA"],
"354 Continue")
- def test_MailFactory_DATA_get_help(self):
+ def test_SMTPIncomingServerFactory_DATA_get_help(self):
"""A DATA command with ``'get help'`` in the email body should
receive::
'250 Delivery in progress'
@@ -687,7 +460,7 @@ class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
"250 Delivery in progress",
noisy=True)
- def test_MailFactory_DATA_get_transport_obfs3(self):
+ def test_SMTPIncomingServerFactory_DATA_get_transport_obfs3(self):
"""A DATA command with ``'get transport obfs3'`` in the email body
should receive::
'250 Delivery in progress'
@@ -701,7 +474,7 @@ class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
"250 Delivery in progress",
noisy=True)
- def test_MailFactory_DATA_To_bridges_plus_zh_CN(self):
+ def test_SMTPIncomingServerFactory_DATA_To_bridges_plus_zh_CN(self):
"""Test sending to 'bridges+zh_CN' address for Chinese translations."""
# TODO: Add tests which use '+' syntax in mailTo in order to test
# email translations. Do this when some strings have been translated.
@@ -716,7 +489,7 @@ class MailFactoryTests(SMTPTestCaseMixin, unittest.TestCase):
"250 Delivery in progress",
noisy=True)
- def test_MailFactory_DATA_get_bridges_QUIT(self):
+ def test_SMTPIncomingServerFactory_DATA_get_bridges_QUIT(self):
"""Test sending 'DATA' with 'get bridges', then sending 'QUIT'."""
emailText = self._buildEmail()
self._test(['HELO localhost',
@@ -732,9 +505,9 @@ class EmailServerServiceTests(SMTPTestCaseMixin, unittest.TestCase):
"""Unittests for :func:`bridgedb.email.server.addServer`."""
def setUp(self):
- """Create a server.MailContext and EmailBasedDistributor."""
+ """Create a server.MailServerContext and EmailBasedDistributor."""
self.config = _createConfig()
- self.context = _createMailContext(self.config)
+ self.context = _createMailServerContext(self.config)
self.smtpFromAddr = self.context.smtpFromAddr # 'bridges at localhost'
self.sched = Unscheduled()
self.dist = self.context.distributor
More information about the tor-commits
mailing list