commit 0556e7daa8504756b34384a0f9b5dfa753556c09 Author: Matthew Finkel Matthew.Finkel@gmail.com Date: Fri Mar 21 03:19:23 2014 +0000
More unit tests for Email Server --- lib/bridgedb/test/test_EmailServer.py | 160 +++++++++++++++++++++++++++++++++ 1 file changed, 160 insertions(+)
diff --git a/lib/bridgedb/test/test_EmailServer.py b/lib/bridgedb/test/test_EmailServer.py index 4c6bcd7..e8b9360 100644 --- a/lib/bridgedb/test/test_EmailServer.py +++ b/lib/bridgedb/test/test_EmailServer.py @@ -17,8 +17,13 @@ import os import shutil
from io import StringIO +import copy
from bridgedb import EmailServer +from bridgedb.Dist import BadEmail +from bridgedb.Dist import EmailBasedDistributor +from bridgedb.EmailServer import MailContext +from bridgedb.Time import NoSchedule from bridgedb.persistent import Conf from bridgedb.test.util import fileCheckDecorator from twisted.python import log @@ -29,8 +34,36 @@ TEST_CONFIG_FILE = StringIO(unicode("""\ EMAIL_DIST = True EMAIL_GPG_SIGNING_ENABLED = True EMAIL_GPG_SIGNING_KEY = 'TESTING.subkeys.sec' +EMAIL_DOMAIN_MAP = {} +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@localhost" +EMAIL_N_BRIDGES_PER_ANSWER = 3 +EMAIL_FROM_ADDR = "bridges@localhost" +EMAIL_BIND_IP = "127.0.0.1" +EMAIL_PORT = 5225 """))
+class FakeDistributor(EmailBasedDistributor): + def __init__(self, key, domainmap, domainrules, answerParameters=None, + bridges=None): + super(FakeDistributor, self).__init__(key, domainmap, domainrules, + answerParameters) + if bridges: + self.bridges = bridges + else: + self.bridges = [] + + def getBridgesForEmail(self, emailaddr, epoch, N=1, + parameters=None, countryCode=None, bridgeFilterRules=None): + return self.bridges[:N] +
class EmailGnuPGTest(unittest.TestCase): """Tests for :func:`bridgedb.EmailServer.getGPGContext`.""" @@ -96,3 +129,130 @@ class EmailGnuPGTest(unittest.TestCase): self.makeBadKey() ctx = EmailServer.getGPGContext(self.config) self.assertTrue(ctx is None) + +class EmailCompositionTests(unittest.TestCase): + """Tests for :func:`bridgedb.EmailServer.getMailResponse`.""" + + def setUp(self): + """Create fake email and associated data""" + configuration = {} + TEST_CONFIG_FILE.seek(0) + compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec') + exec compiled in configuration + self.config = Conf(**configuration) + + # TODO: Add headers if we start validating them + self.lines = ["From: %s@%s.com", "To: %s@example.net", + "Subject: testing", "\n", "get bridges"] + self.distributor = FakeDistributor('key', {}, {}, []) + self.ctx = MailContext(self.config, self.distributor, NoSchedule()) + + def test_getMailResponseNoFrom(self): + lines = self.lines + lines[0] = "" + lines[1] = self.lines[1] % "bridges" + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + + def test_getMailResponseBadAddress(self): + lines = copy.copy(self.lines) + lines[0] = self.lines[0] % ("testing?", "example") + lines[1] = self.lines[1] % "bridges" + lines[2] = "" + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + #lines[0] = self.lines[0] % ("<>>", "example") + lines[0] = "From: %s@%s.com" % ("<>>", "example") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + + def test_getMailResponseInvalidDomain(self): + lines = copy.copy(self.lines) + lines[0] = self.lines[0] % ("testing", "exa#mple") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + lines[0] = self.lines[0] % ("testing", "exam+ple") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + + def test_getMailResponseDKIM(self): + 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.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], None) + self.assertEqual(ret[1], None) + lines[0] = self.lines[0] % ("testing", "example") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], "testing@example.com") + self.assertIsInstance(ret[1], StringIO) + mail = ret[1].getvalue() + self.assertNotEqual(mail.find("no bridges currently"), -1) + + def test_getMailResponseMailContent(self): + lines = copy.copy(self.lines) + lines[0] = self.lines[0] % ("testing", "example") + lines.append("transport obfs") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], "testing@example.com") + self.assertIsInstance(ret[1], StringIO) + mail = ret[1].getvalue() + self.assertNotEqual(mail.find("no bridges currently"), -1) + lines.append("transport obfs") + lines.append("unblocked webz") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], "testing@example.com") + self.assertIsInstance(ret[1], StringIO) + mail = ret[1].getvalue() + self.assertNotEqual(mail.find("no bridges currently"), -1) + lines.append("ipv6") + ret = EmailServer.getMailResponse(lines, self.ctx) + self.assertIsInstance(ret, tuple) + self.assertEqual(len(ret), 2) + self.assertEqual(ret[0], "testing@example.com") + self.assertIsInstance(ret[1], StringIO) + mail = ret[1].getvalue() + self.assertNotEqual(mail.find("no bridges currently"), -1) + +class EmailServerServiceTests(unittest.TestCase): + def setUp(self): + configuration = {} + TEST_CONFIG_FILE.seek(0) + compiled = compile(TEST_CONFIG_FILE.read(), '<string>', 'exec') + exec compiled in configuration + self.config = Conf(**configuration) + + # TODO: Add headers if we start validating them + self.lines = ["From: %s@%s.com", "To: %s@example.net", + "Subject: testing", "\n", "get bridges"] + self.distributor = FakeDistributor('key', {}, {}, []) + self.ctx = MailContext(self.config, self.distributor, NoSchedule()) + + def test_receiveMail(self): + self.skip = True + raise unittest.SkipTest("Not finished yet") + from twisted.internet import reactor + EmailServer.addSMTPServer(self.config, self.distributor, NoSchedule)