[bridgedb/develop] Move test_EmailServer.py → test_email_server.py and update imports.

isis at torproject.org isis at torproject.org
Wed May 14 03:42:27 UTC 2014


commit 3176bcd78da2515af4834c925689609e5bba015d
Author: Isis Lovecruft <isis at torproject.org>
Date:   Mon May 5 19:57:59 2014 +0000

    Move test_EmailServer.py → test_email_server.py and update imports.
---
 lib/bridgedb/test/test_EmailServer.py  |  344 --------------------------------
 lib/bridgedb/test/test_email_server.py |  341 +++++++++++++++++++++++++++++++
 2 files changed, 341 insertions(+), 344 deletions(-)

diff --git a/lib/bridgedb/test/test_EmailServer.py b/lib/bridgedb/test/test_EmailServer.py
deleted file mode 100644
index 2ecc7f9..0000000
--- a/lib/bridgedb/test/test_EmailServer.py
+++ /dev/null
@@ -1,344 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# This file is part of BridgeDB, a Tor bridge distribution system.
-#
-# :authors: Isis Lovecruft 0xA3ADB67A2CDB8B35 <isis at torproject.org>
-#           please also see AUTHORS file
-# :copyright: (c) 2013, Isis Lovecruft
-#             (c) 2007-2013, The Tor Project, Inc.
-#             (c) 2007-2013, all entities within the AUTHORS file
-# :license: 3-Clause BSD, see LICENSE for licensing information
-
-"""Unittests for the :mod:`bridgedb.EmailServer` module."""
-
-from __future__ import print_function
-
-import os
-import shutil
-
-import io
-import copy
-
-from bridgedb import EmailServer
-from bridgedb.Dist import EmailBasedDistributor
-from bridgedb.EmailServer import MailContext
-from bridgedb.Time import NoSchedule
-from bridgedb.parse.addr import BadEmail
-from bridgedb.persistent import Conf
-from bridgedb.test.test_HTTPServer import DummyBridge
-from bridgedb.test.util import fileCheckDecorator
-
-from twisted.python import log
-from twisted.internet import defer
-from twisted.trial import unittest
-
-
-TEST_CONFIG_FILE = io.StringIO(unicode("""\
-EMAIL_DIST = True
-EMAIL_INCLUDE_FINGERPRINTS = True
-EMAIL_GPG_SIGNING_ENABLED = True
-EMAIL_GPG_SIGNING_KEY = 'TESTING.subkeys.sec'
-EMAIL_DOMAIN_MAP = {
-   'googlemail.com': 'gmail.com',
-   'mail.google.com': 'gmail.com',
-}
-EMAIL_DOMAIN_RULES = {
-   'gmail.com': ["ignore_dots", "dkim"],
-   'example.com': [],
-}
-EMAIL_DOMAINS = ["gmail.com", "example.com"]
-EMAIL_USERNAME = "bridges"
-EMAIL_SMTP_HOST = "127.0.0.1"
-EMAIL_SMTP_PORT = 25
-EMAIL_SMTP_FROM_ADDR = "bridges at localhost"
-EMAIL_N_BRIDGES_PER_ANSWER = 3
-EMAIL_FROM_ADDR = "bridges at localhost"
-EMAIL_BIND_IP = "127.0.0.1"
-EMAIL_PORT = 5225
-"""))
-
-def _createMailContext(distributor=None):
-    configuration = {}
-    TEST_CONFIG_FILE.seek(0)
-    compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec')
-    exec compiled in configuration
-    config = Conf(**configuration)
-
-    if not distributor:
-        distributor = DummyEmailDistributor(
-            domainmap=config.EMAIL_DOMAIN_MAP,
-            domainrules=config.EMAIL_DOMAIN_RULES)
-
-    ctx = MailContext(config, distributor, NoSchedule())
-    return ctx
-
-
-class DummyEmailDistributor(object):
-    """A mocked :class:`bridgedb.Dist.EmailBasedDistributor` which is used to
-    test :class:`bridgedb.EmailServer`.
-    """
-
-    def __init__(self, key=None, domainmap=None, domainrules=None,
-                 answerParameters=None):
-        """None of the parameters are really used, except ``ctx`` ― they are
-        just there to retain an identical method signature.
-        """
-        self.key = self.__class__.__name__
-        self.domainmap = domainmap
-        self.domainrules = domainrules
-        self.answerParameters = answerParameters
-
-    def getBridgesForEmail(self, emailaddress, epoch, N=1, parameters=None,
-                           countryCode=None, bridgeFilterRules=None):
-        """Needed because it's called in
-        :meth:`WebResourceBridges.getBridgesForIP`.
-        """
-        return [DummyBridge() for _ in xrange(N)]
-
-
-class EmailGnuPGTest(unittest.TestCase):
-    """Tests for :func:`bridgedb.EmailServer.getGPGContext`."""
-
-    timeout = 15
-
-    @fileCheckDecorator
-    def doCopyFile(self, src, dst, description=None):
-        shutil.copy(src, dst)
-
-    def removeRundir(self):
-        if os.path.isdir(self.runDir):
-            shutil.rmtree(self.runDir)
-
-    def makeBadKey(self):
-        keyfile = os.path.join(self.runDir, 'badkey.asc')
-        with open(keyfile, 'wb') as badkey:
-            badkey.write('NO PASARÁN, DEATH CAKES!')
-            badkey.flush()
-        self.setKey(keyfile)
-
-    def setKey(self, keyfile=''):
-        setattr(self.config, 'EMAIL_GPG_SIGNING_KEY', keyfile)
-
-    def setUp(self):
-        here          = os.getcwd()
-        topDir        = here.rstrip('_trial_temp')
-        self.runDir   = os.path.join(here, 'rundir')
-        self.gpgFile  = os.path.join(topDir, 'gnupghome', 'TESTING.subkeys.sec')
-        self.gpgMoved = os.path.join(here, 'TESTING.subkeys.sec')
-
-        if not os.path.isdir(self.runDir):
-            os.makedirs(self.runDir)
-
-        configuration = {}
-        TEST_CONFIG_FILE.seek(0)
-        compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec')
-        exec compiled in configuration
-        self.config = Conf(**configuration)
-
-        self.addCleanup(self.removeRundir)
-
-    def test_getGPGContext_good_keyfile(self):
-        """Test EmailServer.getGPGContext() with a good key filename.
-
-        XXX: See #5463.
-        """
-        raise unittest.SkipTest(
-            "See #5463 for why this test fails when it should pass")
-
-        self.doCopyFile(self.gpgFile, self.gpgMoved, "GnuPG test keyfile")
-        ctx = EmailServer.getGPGContext(self.config)
-        self.assertIsInstance(ctx, EmailServer.gpgme.Context)
-
-    def test_getGPGContext_missing_keyfile(self):
-        """Test EmailServer.getGPGContext() with a missing key filename."""
-        self.setKey('missing-keyfile.asc')
-        ctx = EmailServer.getGPGContext(self.config)
-        self.assertTrue(ctx is None)
-
-    def test_getGPGContext_bad_keyfile(self):
-        """Test EmailServer.getGPGContext() with a missing key filename."""
-        self.makeBadKey()
-        ctx = EmailServer.getGPGContext(self.config)
-        self.assertTrue(ctx is None)
-
-
-class EmailResponseTests(unittest.TestCase):
-    """Tests for :func:`bridgedb.EmailServer.getMailResponse`."""
-
-    def setUp(self):
-        """Create fake email, distributor, and associated context data."""
-        # TODO: Add headers if we start validating them
-        self.lines = ["From: %s@%s.com",
-                      "To: bridges at localhost",
-                      "Subject: testing",
-                      "",
-                      "get bridges"]
-        self.ctx = _createMailContext()
-
-    def _isTwoTupleOfNone(self, reply):
-        """Check that a return value is ``(None, None)``."""
-        self.assertIsInstance(reply, tuple)
-        self.assertEqual(len(reply), 2)
-        self.assertEqual(reply[0], None)
-        self.assertEqual(reply[1], None)
-
-    def _isTwoTupleOfAddrAndClass(self, reply, address="testing at localhost",
-                                  klass=io.StringIO):
-        self.assertIsInstance(reply, tuple)
-        self.assertEqual(len(reply), 2)
-        self.assertEqual(reply[0], address)
-        self.assertIsInstance(reply[1], klass)
-
-    def test_getMailResponse_noFrom(self):
-        """A received email without a "From:" or "Sender:" header shouldn't
-        receive a response.
-        """
-        lines = self.lines
-        lines[0] = ""
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_badAddress(self):
-        """Don't respond to RFC2822 malformed source addresses."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing*.?\"", "example")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_anotherBadAddress(self):
-        """Don't respond to RFC2822 malformed source addresses."""
-        lines = copy.copy(self.lines)
-        lines[0] = "From: Mallory %s@%s.com" % ("<>>", "example")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_invalidDomain(self):
-        """Don't respond to RFC2822 malformed source addresses."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "exa#mple")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_anotherInvalidDomain(self):
-        """Don't respond to RFC2822 malformed source addresses."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "exam+ple")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_DKIM_badDKIMheader(self):
-        """An email with an 'X-DKIM-Authentication-Result:' header appended
-        after the body should not receive a response.
-        """
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "gmail")
-        lines.append("X-DKIM-Authentication-Result: ")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self._isTwoTupleOfNone(ret)
-
-    def test_getMailResponse_DKIM(self):
-        """An email with a good DKIM header should be responded to."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "localhost")
-        lines.insert(3, "X-DKIM-Authentication-Result: ")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self.skip = True
-        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
-                                " the email distributor should pass these"\
-                                " tests.")
-        self._isTwoTupleOfAddrAndClass(ret)
-        mail = ret[1].getvalue()
-        self.assertEqual(mail.find("no bridges currently"), -1)
-
-    def test_getMailResponse_bridges_obfs3(self):
-        """A request for 'transport obfs3' should receive a response."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "localhost")
-        lines[4] = "transport obfs3"
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self.skip = True
-        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
-                                " the email distributor should pass these"\
-                                " tests.")
-        self._isTwoTupleOfAddrAndClass(ret)
-        mail = ret[1].getvalue()
-        self.assertEqual(mail.find("no bridges currently"), -1)
-
-    def test_getMailResponse_bridges_obfsobfswebz(self):
-        """We should only pay attention to the *last* in a crazy request."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "localhost")
-        lines[4] = "unblocked webz"
-        lines.append("transport obfs2")
-        lines.append("transport obfs3")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self.skip = True
-        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
-                                " the email distributor should pass these"\
-                                " tests.")
-        self._isTwoTupleOfAddrAndClass(ret)
-        mail = ret[1].getvalue()
-        self.assertNotEqual(mail.find("no bridges currently"), -1)
-
-    def test_getMailResponse_bridges_obfsobfswebzipv6(self):
-        """We should *still* only pay attention to the *last* request."""
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "localhost")
-        lines[4] = "transport obfs3"
-        lines.append("unblocked webz")
-        lines.append("ipv6")
-        lines.append("transport obfs2")
-        ret = EmailServer.getMailResponse(lines, self.ctx)
-        self.skip = True
-        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
-                                " the email distributor should pass these"\
-                                " tests.")
-        self._isTwoTupleOfAddrAndClass(ret)
-        mail = ret[1].getvalue()
-        self.assertNotEqual(mail.find("no bridges currently"), -1)
-
-
-class EmailReplyTests(unittest.TestCase):
-    """Tests for ``EmailServer.replyToMail()``."""
-
-    def setUp(self):
-        """Create fake email, distributor, and associated context data."""
-        # TODO: Add headers if we start validating them
-        self.lines = ["From: %s@%s.com",
-                      "To: bridges at example.net",
-                      "Subject: testing",
-                      "\n",
-                      "get bridges"]
-        self.ctx = _createMailContext()
-
-    def test_replyToMail(self):
-        self.skip = True
-        raise unittest.SkipTest("We'll have to fake the EmailServer for this one,"\
-                                " it requires a TCP connection to localhost.")
-
-        def callback(reply):
-            self.assertSubstring("Here are your bridges", reply)
-
-        lines = copy.copy(self.lines)
-        lines[0] = self.lines[0] % ("testing", "example")
-        reply = EmailServer.replyToMail(lines, self.ctx)
-
-        self.assertIsInstance(reply, defer.Deferred)
-
-        reply.addCallback(callback)
-        return reply
-
-
-class EmailServerServiceTests(unittest.TestCase):
-    def setUp(self):
-        # TODO: Add headers if we start validating them
-        self.lines = ["From: %s@%s.com", "To: %s at example.net",
-                      "Subject: testing", "\n", "get bridges"]
-        self.distributor = DummyEmailDistributor('key', {}, {}, [])
-        self.ctx = _createMailContext(self.distributor)
-
-    def test_receiveMail(self):
-        self.skip = True
-        raise unittest.SkipTest("Not finished yet")
-        from twisted.internet import reactor
-        EmailServer.addSMTPServer(self.ctx.cfg, self.distributor, NoSchedule)
diff --git a/lib/bridgedb/test/test_email_server.py b/lib/bridgedb/test/test_email_server.py
new file mode 100644
index 0000000..e925f07
--- /dev/null
+++ b/lib/bridgedb/test/test_email_server.py
@@ -0,0 +1,341 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of BridgeDB, a Tor bridge distribution system.
+#
+# :authors: Isis Lovecruft 0xA3ADB67A2CDB8B35 <isis at torproject.org>
+#           please also see AUTHORS file
+# :copyright: (c) 2013, Isis Lovecruft
+#             (c) 2007-2013, The Tor Project, Inc.
+#             (c) 2007-2013, all entities within the AUTHORS file
+# :license: 3-Clause BSD, see LICENSE for licensing information
+
+"""Unittests for the :mod:`bridgedb.email.server` module."""
+
+from __future__ import print_function
+
+import os
+import shutil
+
+import io
+import copy
+
+from bridgedb.Dist import EmailBasedDistributor
+from bridgedb.email import server
+from bridgedb.email.server import MailContext
+from bridgedb.Time import NoSchedule
+from bridgedb.parse.addr import BadEmail
+from bridgedb.persistent import Conf
+from bridgedb.test.test_HTTPServer import DummyBridge
+from bridgedb.test.util import fileCheckDecorator
+
+from twisted.python import log
+from twisted.internet import defer
+from twisted.trial import unittest
+
+
+TEST_CONFIG_FILE = io.StringIO(unicode("""\
+EMAIL_DIST = True
+EMAIL_INCLUDE_FINGERPRINTS = True
+EMAIL_GPG_SIGNING_ENABLED = True
+EMAIL_GPG_SIGNING_KEY = 'TESTING.subkeys.sec'
+EMAIL_DOMAIN_MAP = {
+   'googlemail.com': 'gmail.com',
+   'mail.google.com': 'gmail.com',
+}
+EMAIL_DOMAIN_RULES = {
+   'gmail.com': ["ignore_dots", "dkim"],
+   'example.com': [],
+}
+EMAIL_DOMAINS = ["gmail.com", "example.com"]
+EMAIL_USERNAME = "bridges"
+EMAIL_SMTP_HOST = "127.0.0.1"
+EMAIL_SMTP_PORT = 25
+EMAIL_SMTP_FROM_ADDR = "bridges at localhost"
+EMAIL_N_BRIDGES_PER_ANSWER = 3
+EMAIL_FROM_ADDR = "bridges at localhost"
+EMAIL_BIND_IP = "127.0.0.1"
+EMAIL_PORT = 5225
+"""))
+
+def _createMailContext(distributor=None):
+    configuration = {}
+    TEST_CONFIG_FILE.seek(0)
+    compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec')
+    exec compiled in configuration
+    config = Conf(**configuration)
+
+    if not distributor:
+        distributor = DummyEmailDistributor(
+            domainmap=config.EMAIL_DOMAIN_MAP,
+            domainrules=config.EMAIL_DOMAIN_RULES)
+
+    ctx = MailContext(config, distributor, NoSchedule())
+    return ctx
+
+
+class DummyEmailDistributor(object):
+    """A mocked :class:`bridgedb.Dist.EmailBasedDistributor` which is used to
+    test :class:`bridgedb.EmailServer`.
+    """
+
+    def __init__(self, key=None, domainmap=None, domainrules=None,
+                 answerParameters=None):
+        """None of the parameters are really used, ― they are just there to retain an
+        identical method signature.
+        """
+        self.key = self.__class__.__name__
+        self.domainmap = domainmap
+        self.domainrules = domainrules
+        self.answerParameters = answerParameters
+
+    def getBridgesForEmail(self, emailaddress, epoch, N=1, parameters=None,
+                           countryCode=None, bridgeFilterRules=None):
+        return [DummyBridge() for _ in xrange(N)]
+
+
+class EmailGnuPGTest(unittest.TestCase):
+    """Tests for :func:`bridgedb.EmailServer.getGPGContext`."""
+
+    timeout = 15
+
+    @fileCheckDecorator
+    def doCopyFile(self, src, dst, description=None):
+        shutil.copy(src, dst)
+
+    def removeRundir(self):
+        if os.path.isdir(self.runDir):
+            shutil.rmtree(self.runDir)
+
+    def makeBadKey(self):
+        keyfile = os.path.join(self.runDir, 'badkey.asc')
+        with open(keyfile, 'wb') as badkey:
+            badkey.write('NO PASARÁN, DEATH CAKES!')
+            badkey.flush()
+        self.setKey(keyfile)
+
+    def setKey(self, keyfile=''):
+        setattr(self.config, 'EMAIL_GPG_SIGNING_KEY', keyfile)
+
+    def setUp(self):
+        here          = os.getcwd()
+        topDir        = here.rstrip('_trial_temp')
+        self.runDir   = os.path.join(here, 'rundir')
+        self.gpgFile  = os.path.join(topDir, 'gnupghome', 'TESTING.subkeys.sec')
+        self.gpgMoved = os.path.join(here, 'TESTING.subkeys.sec')
+
+        if not os.path.isdir(self.runDir):
+            os.makedirs(self.runDir)
+
+        configuration = {}
+        TEST_CONFIG_FILE.seek(0)
+        compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec')
+        exec compiled in configuration
+        self.config = Conf(**configuration)
+
+        self.addCleanup(self.removeRundir)
+
+    def test_getGPGContext_good_keyfile(self):
+        """Test EmailServer.getGPGContext() with a good key filename.
+
+        XXX: See #5463.
+        """
+        raise unittest.SkipTest(
+            "See #5463 for why this test fails when it should pass")
+
+        self.doCopyFile(self.gpgFile, self.gpgMoved, "GnuPG test keyfile")
+        ctx = EmailServer.getGPGContext(self.config)
+        self.assertIsInstance(ctx, EmailServer.gpgme.Context)
+
+    def test_getGPGContext_missing_keyfile(self):
+        """Test EmailServer.getGPGContext() with a missing key filename."""
+        self.setKey('missing-keyfile.asc')
+        ctx = EmailServer.getGPGContext(self.config)
+        self.assertTrue(ctx is None)
+
+    def test_getGPGContext_bad_keyfile(self):
+        """Test EmailServer.getGPGContext() with a missing key filename."""
+        self.makeBadKey()
+        ctx = EmailServer.getGPGContext(self.config)
+        self.assertTrue(ctx is None)
+
+
+class EmailResponseTests(unittest.TestCase):
+    """Tests for :func:`bridgedb.EmailServer.getMailResponse`."""
+
+    def setUp(self):
+        """Create fake email, distributor, and associated context data."""
+        # TODO: Add headers if we start validating them
+        self.lines = ["From: %s@%s.com",
+                      "To: bridges at localhost",
+                      "Subject: testing",
+                      "",
+                      "get bridges"]
+        self.ctx = _createMailContext()
+
+    def _isTwoTupleOfNone(self, reply):
+        """Check that a return value is ``(None, None)``."""
+        self.assertIsInstance(reply, tuple)
+        self.assertEqual(len(reply), 2)
+        self.assertEqual(reply[0], None)
+        self.assertEqual(reply[1], None)
+
+    def _isTwoTupleOfAddrAndClass(self, reply, address="testing at localhost",
+                                  klass=io.StringIO):
+        self.assertIsInstance(reply, tuple)
+        self.assertEqual(len(reply), 2)
+        self.assertEqual(reply[0], address)
+        self.assertIsInstance(reply[1], klass)
+
+    def test_getMailResponse_noFrom(self):
+        """A received email without a "From:" or "Sender:" header shouldn't
+        receive a response.
+        """
+        lines = self.lines
+        lines[0] = ""
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_badAddress(self):
+        """Don't respond to RFC2822 malformed source addresses."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing*.?\"", "example")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_anotherBadAddress(self):
+        """Don't respond to RFC2822 malformed source addresses."""
+        lines = copy.copy(self.lines)
+        lines[0] = "From: Mallory %s@%s.com" % ("<>>", "example")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_invalidDomain(self):
+        """Don't respond to RFC2822 malformed source addresses."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "exa#mple")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_anotherInvalidDomain(self):
+        """Don't respond to RFC2822 malformed source addresses."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "exam+ple")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_DKIM_badDKIMheader(self):
+        """An email with an 'X-DKIM-Authentication-Result:' header appended
+        after the body should not receive a response.
+        """
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "gmail")
+        lines.append("X-DKIM-Authentication-Result: ")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self._isTwoTupleOfNone(ret)
+
+    def test_getMailResponse_DKIM(self):
+        """An email with a good DKIM header should be responded to."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "localhost")
+        lines.insert(3, "X-DKIM-Authentication-Result: ")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self.skip = True
+        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
+                                " the email distributor should pass these"\
+                                " tests.")
+        self._isTwoTupleOfAddrAndClass(ret)
+        mail = ret[1].getvalue()
+        self.assertEqual(mail.find("no bridges currently"), -1)
+
+    def test_getMailResponse_bridges_obfs3(self):
+        """A request for 'transport obfs3' should receive a response."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "localhost")
+        lines[4] = "transport obfs3"
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self.skip = True
+        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
+                                " the email distributor should pass these"\
+                                " tests.")
+        self._isTwoTupleOfAddrAndClass(ret)
+        mail = ret[1].getvalue()
+        self.assertEqual(mail.find("no bridges currently"), -1)
+
+    def test_getMailResponse_bridges_obfsobfswebz(self):
+        """We should only pay attention to the *last* in a crazy request."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "localhost")
+        lines[4] = "unblocked webz"
+        lines.append("transport obfs2")
+        lines.append("transport obfs3")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self.skip = True
+        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
+                                " the email distributor should pass these"\
+                                " tests.")
+        self._isTwoTupleOfAddrAndClass(ret)
+        mail = ret[1].getvalue()
+        self.assertNotEqual(mail.find("no bridges currently"), -1)
+
+    def test_getMailResponse_bridges_obfsobfswebzipv6(self):
+        """We should *still* only pay attention to the *last* request."""
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "localhost")
+        lines[4] = "transport obfs3"
+        lines.append("unblocked webz")
+        lines.append("ipv6")
+        lines.append("transport obfs2")
+        ret = EmailServer.getMailResponse(lines, self.ctx)
+        self.skip = True
+        raise unittest.SkipTest("Broken; not sure why. Manual testing says"\
+                                " the email distributor should pass these"\
+                                " tests.")
+        self._isTwoTupleOfAddrAndClass(ret)
+        mail = ret[1].getvalue()
+        self.assertNotEqual(mail.find("no bridges currently"), -1)
+
+
+class EmailReplyTests(unittest.TestCase):
+    """Tests for ``EmailServer.replyToMail()``."""
+
+    def setUp(self):
+        """Create fake email, distributor, and associated context data."""
+        # TODO: Add headers if we start validating them
+        self.lines = ["From: %s@%s.com",
+                      "To: bridges at example.net",
+                      "Subject: testing",
+                      "\n",
+                      "get bridges"]
+        self.ctx = _createMailContext()
+
+    def test_replyToMail(self):
+        self.skip = True
+        raise unittest.SkipTest("We'll have to fake the EmailServer for this one,"\
+                                " it requires a TCP connection to localhost.")
+
+        def callback(reply):
+            self.assertSubstring("Here are your bridges", reply)
+
+        lines = copy.copy(self.lines)
+        lines[0] = self.lines[0] % ("testing", "example")
+        reply = EmailServer.replyToMail(lines, self.ctx)
+
+        self.assertIsInstance(reply, defer.Deferred)
+
+        reply.addCallback(callback)
+        return reply
+
+
+class EmailServerServiceTests(unittest.TestCase):
+    def setUp(self):
+        # TODO: Add headers if we start validating them
+        self.lines = ["From: %s@%s.com", "To: %s at example.net",
+                      "Subject: testing", "\n", "get bridges"]
+        self.distributor = DummyEmailDistributor('key', {}, {}, [])
+        self.ctx = _createMailContext(self.distributor)
+
+    def test_receiveMail(self):
+        self.skip = True
+        raise unittest.SkipTest("Not finished yet")
+        from twisted.internet import reactor
+        EmailServer.addSMTPServer(self.ctx.cfg, self.distributor, NoSchedule)





More information about the tor-commits mailing list