1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for large portions of L{twisted.mail}.
17 from zope.interface import Interface, implements
19 from twisted.trial import unittest
20 from twisted.mail import smtp
21 from twisted.mail import pop3
22 from twisted.names import dns
23 from twisted.internet import protocol
24 from twisted.internet import defer
25 from twisted.internet.defer import Deferred
26 from twisted.internet import reactor
27 from twisted.internet import interfaces
28 from twisted.internet import task
29 from twisted.internet.error import DNSLookupError, CannotListenError
30 from twisted.internet.error import ProcessDone, ProcessTerminated
31 from twisted.internet import address
32 from twisted.python import failure
33 from twisted.python.filepath import FilePath
34 from twisted.python.hashlib import md5
36 from twisted import mail
37 import twisted.mail.mail
38 import twisted.mail.maildir
39 import twisted.mail.relay
40 import twisted.mail.relaymanager
41 import twisted.mail.protocols
42 import twisted.mail.alias
44 from twisted.names.error import DNSNameError
45 from twisted.names.dns import RRHeader, Record_CNAME, Record_MX
47 from twisted import cred
48 import twisted.cred.credentials
49 import twisted.cred.checkers
50 import twisted.cred.portal
52 from twisted.test.proto_helpers import LineSendingProtocol
54 class DomainWithDefaultsTestCase(unittest.TestCase):
55 def testMethods(self):
56 d = dict([(x, x + 10) for x in range(10)])
57 d = mail.mail.DomainWithDefaultDict(d, 'Default')
59 self.assertEqual(len(d), 10)
60 self.assertEqual(list(iter(d)), range(10))
61 self.assertEqual(list(d.iterkeys()), list(iter(d)))
63 items = list(d.iteritems())
65 self.assertEqual(items, [(x, x + 10) for x in range(10)])
67 values = list(d.itervalues())
69 self.assertEqual(values, range(10, 20))
73 self.assertEqual(items, [(x, x + 10) for x in range(10)])
77 self.assertEqual(values, range(10, 20))
80 self.assertEqual(d[x], x + 10)
81 self.assertEqual(d.get(x), x + 10)
82 self.failUnless(x in d)
83 self.failUnless(d.has_key(x))
87 self.assertEqual(len(d), 7)
88 self.assertEqual(d[2], 'Default')
89 self.assertEqual(d[4], 'Default')
90 self.assertEqual(d[6], 'Default')
92 d.update({'a': None, 'b': (), 'c': '*'})
93 self.assertEqual(len(d), 10)
94 self.assertEqual(d['a'], None)
95 self.assertEqual(d['b'], ())
96 self.assertEqual(d['c'], '*')
99 self.assertEqual(len(d), 0)
101 self.assertEqual(d.setdefault('key', 'value'), 'value')
102 self.assertEqual(d['key'], 'value')
104 self.assertEqual(d.popitem(), ('key', 'value'))
105 self.assertEqual(len(d), 0)
108 self.assertEqual(d.domains, dcopy.domains)
109 self.assertEqual(d.default, dcopy.default)
112 def _stringificationTest(self, stringifier):
114 Assert that the class name of a L{mail.mail.DomainWithDefaultDict}
115 instance and the string-formatted underlying domain dictionary both
116 appear in the string produced by the given string-returning function.
118 @type stringifier: one-argument callable
119 @param stringifier: either C{str} or C{repr}, to be used to get a
120 string to make assertions against.
122 domain = mail.mail.DomainWithDefaultDict({}, 'Default')
123 self.assertIn(domain.__class__.__name__, stringifier(domain))
124 domain['key'] = 'value'
125 self.assertIn(str({'key': 'value'}), stringifier(domain))
130 L{DomainWithDefaultDict.__str__} should return a string including
131 the class name and the domain mapping held by the instance.
133 self._stringificationTest(str)
138 L{DomainWithDefaultDict.__repr__} should return a string including
139 the class name and the domain mapping held by the instance.
141 self._stringificationTest(repr)
145 class BounceTestCase(unittest.TestCase):
147 self.domain = mail.mail.BounceDomain()
149 def testExists(self):
150 self.assertRaises(smtp.AddressError, self.domain.exists, "any user")
154 self.domain.willRelay("random q emailer", "protocol"),
158 def testMessage(self):
159 self.assertRaises(NotImplementedError, self.domain.startMessage, "whomever")
161 def testAddUser(self):
162 self.domain.addUser("bob", "password")
163 self.assertRaises(smtp.SMTPBadRcpt, self.domain.exists, "bob")
165 class FileMessageTestCase(unittest.TestCase):
167 self.name = "fileMessage.testFile"
168 self.final = "final.fileMessage.testFile"
169 self.f = file(self.name, 'w')
170 self.fp = mail.mail.FileMessage(self.f, self.name, self.final)
182 os.remove(self.final)
186 def testFinalName(self):
187 return self.fp.eomReceived().addCallback(self._cbFinalName)
189 def _cbFinalName(self, result):
190 self.assertEqual(result, self.final)
191 self.failUnless(self.f.closed)
192 self.failIf(os.path.exists(self.name))
194 def testContents(self):
195 contents = "first line\nsecond line\nthird line\n"
196 for line in contents.splitlines():
197 self.fp.lineReceived(line)
198 self.fp.eomReceived()
199 self.assertEqual(file(self.final).read(), contents)
201 def testInterrupted(self):
202 contents = "first line\nsecond line\n"
203 for line in contents.splitlines():
204 self.fp.lineReceived(line)
205 self.fp.connectionLost()
206 self.failIf(os.path.exists(self.name))
207 self.failIf(os.path.exists(self.final))
209 class MailServiceTestCase(unittest.TestCase):
211 self.service = mail.mail.MailService()
213 def testFactories(self):
214 f = self.service.getPOP3Factory()
215 self.failUnless(isinstance(f, protocol.ServerFactory))
216 self.failUnless(f.buildProtocol(('127.0.0.1', 12345)), pop3.POP3)
218 f = self.service.getSMTPFactory()
219 self.failUnless(isinstance(f, protocol.ServerFactory))
220 self.failUnless(f.buildProtocol(('127.0.0.1', 12345)), smtp.SMTP)
222 f = self.service.getESMTPFactory()
223 self.failUnless(isinstance(f, protocol.ServerFactory))
224 self.failUnless(f.buildProtocol(('127.0.0.1', 12345)), smtp.ESMTP)
226 def testPortals(self):
229 self.service.portals['domain'] = o1
230 self.service.portals[''] = o2
232 self.failUnless(self.service.lookupPortal('domain') is o1)
233 self.failUnless(self.service.defaultPortal() is o2)
236 class StringListMailboxTests(unittest.TestCase):
238 Tests for L{StringListMailbox}, an in-memory only implementation of
241 def test_listOneMessage(self):
243 L{StringListMailbox.listMessages} returns the length of the message at
244 the offset into the mailbox passed to it.
246 mailbox = mail.maildir.StringListMailbox(["abc", "ab", "a"])
247 self.assertEqual(mailbox.listMessages(0), 3)
248 self.assertEqual(mailbox.listMessages(1), 2)
249 self.assertEqual(mailbox.listMessages(2), 1)
252 def test_listAllMessages(self):
254 L{StringListMailbox.listMessages} returns a list of the lengths of all
255 messages if not passed an index.
257 mailbox = mail.maildir.StringListMailbox(["a", "abc", "ab"])
258 self.assertEqual(mailbox.listMessages(), [1, 3, 2])
261 def test_getMessage(self):
263 L{StringListMailbox.getMessage} returns a file-like object from which
264 the contents of the message at the given offset into the mailbox can be
267 mailbox = mail.maildir.StringListMailbox(["foo", "real contents"])
268 self.assertEqual(mailbox.getMessage(1).read(), "real contents")
271 def test_getUidl(self):
273 L{StringListMailbox.getUidl} returns a unique identifier for the
274 message at the given offset into the mailbox.
276 mailbox = mail.maildir.StringListMailbox(["foo", "bar"])
277 self.assertNotEqual(mailbox.getUidl(0), mailbox.getUidl(1))
280 def test_deleteMessage(self):
282 L{StringListMailbox.deleteMessage} marks a message for deletion causing
283 further requests for its length to return 0.
285 mailbox = mail.maildir.StringListMailbox(["foo"])
286 mailbox.deleteMessage(0)
287 self.assertEqual(mailbox.listMessages(0), 0)
288 self.assertEqual(mailbox.listMessages(), [0])
291 def test_undeleteMessages(self):
293 L{StringListMailbox.undeleteMessages} causes any messages marked for
294 deletion to be returned to their original state.
296 mailbox = mail.maildir.StringListMailbox(["foo"])
297 mailbox.deleteMessage(0)
298 mailbox.undeleteMessages()
299 self.assertEqual(mailbox.listMessages(0), 3)
300 self.assertEqual(mailbox.listMessages(), [3])
305 L{StringListMailbox.sync} causes any messages as marked for deletion to
306 be permanently deleted.
308 mailbox = mail.maildir.StringListMailbox(["foo"])
309 mailbox.deleteMessage(0)
311 mailbox.undeleteMessages()
312 self.assertEqual(mailbox.listMessages(0), 0)
313 self.assertEqual(mailbox.listMessages(), [0])
317 class FailingMaildirMailboxAppendMessageTask(mail.maildir._MaildirMailboxAppendMessageTask):
321 def osopen(self, fn, attr, mode):
323 return os.open(fn, attr, mode)
325 raise OSError(errno.EPERM, "Faked Permission Problem")
326 def oswrite(self, fh, data):
328 return os.write(fh, data)
330 raise OSError(errno.ENOSPC, "Faked Space problem")
331 def osrename(self, oldname, newname):
332 if self._renamestate:
333 return os.rename(oldname, newname)
335 raise OSError(errno.EPERM, "Faked Permission Problem")
338 class _AppendTestMixin(object):
340 Mixin for L{MaildirMailbox.appendMessage} test cases which defines a helper
341 for serially appending multiple messages to a mailbox.
343 def _appendMessages(self, mbox, messages):
345 Deliver the given messages one at a time. Delivery is serialized to
346 guarantee a predictable order in the mailbox (overlapped message deliver
347 makes no guarantees about which message which appear first).
352 d = mbox.appendMessage(m)
353 d.addCallback(results.append)
355 d = task.cooperate(append()).whenDone()
356 d.addCallback(lambda ignored: results)
361 class MaildirAppendStringTestCase(unittest.TestCase, _AppendTestMixin):
363 Tests for L{MaildirMailbox.appendMessage} when invoked with a C{str}.
366 self.d = self.mktemp()
367 mail.maildir.initializeMaildir(self.d)
370 def _append(self, ignored, mbox):
371 d = mbox.appendMessage('TEST')
372 return self.assertFailure(d, Exception)
375 def _setState(self, ignored, mbox, rename=None, write=None, open=None):
377 Change the behavior of future C{rename}, C{write}, or C{open} calls made
378 by the mailbox C{mbox}.
380 @param rename: If not C{None}, a new value for the C{_renamestate}
381 attribute of the mailbox's append factory. The original value will
382 be restored at the end of the test.
384 @param write: Like C{rename}, but for the C{_writestate} attribute.
386 @param open: Like C{rename}, but for the C{_openstate} attribute.
388 if rename is not None:
390 setattr, mbox.AppendFactory, '_renamestate',
391 mbox.AppendFactory._renamestate)
392 mbox.AppendFactory._renamestate = rename
393 if write is not None:
395 setattr, mbox.AppendFactory, '_writestate',
396 mbox.AppendFactory._writestate)
397 mbox.AppendFactory._writestate = write
400 setattr, mbox.AppendFactory, '_openstate',
401 mbox.AppendFactory._openstate)
402 mbox.AppendFactory._openstate = open
405 def test_append(self):
407 L{MaildirMailbox.appendMessage} returns a L{Deferred} which fires when
408 the message has been added to the end of the mailbox.
410 mbox = mail.maildir.MaildirMailbox(self.d)
411 mbox.AppendFactory = FailingMaildirMailboxAppendMessageTask
413 d = self._appendMessages(mbox, ["X" * i for i in range(1, 11)])
414 d.addCallback(self.assertEqual, [None] * 10)
415 d.addCallback(self._cbTestAppend, mbox)
419 def _cbTestAppend(self, ignored, mbox):
421 Check that the mailbox has the expected number (ten) of messages in it,
422 and that each has the expected contents, and that they are in the same
423 order as that in which they were appended.
425 self.assertEqual(len(mbox.listMessages()), 10)
427 [len(mbox.getMessage(i).read()) for i in range(10)],
429 # test in the right order: last to first error location.
430 self._setState(None, mbox, rename=False)
431 d = self._append(None, mbox)
432 d.addCallback(self._setState, mbox, rename=True, write=False)
433 d.addCallback(self._append, mbox)
434 d.addCallback(self._setState, mbox, write=True, open=False)
435 d.addCallback(self._append, mbox)
436 d.addCallback(self._setState, mbox, open=True)
441 class MaildirAppendFileTestCase(unittest.TestCase, _AppendTestMixin):
443 Tests for L{MaildirMailbox.appendMessage} when invoked with a C{str}.
446 self.d = self.mktemp()
447 mail.maildir.initializeMaildir(self.d)
450 def test_append(self):
452 L{MaildirMailbox.appendMessage} returns a L{Deferred} which fires when
453 the message has been added to the end of the mailbox.
455 mbox = mail.maildir.MaildirMailbox(self.d)
457 for i in xrange(1, 11):
458 temp = tempfile.TemporaryFile()
461 messages.append(temp)
462 self.addCleanup(temp.close)
464 d = self._appendMessages(mbox, messages)
465 d.addCallback(self._cbTestAppend, mbox)
469 def _cbTestAppend(self, result, mbox):
471 Check that the mailbox has the expected number (ten) of messages in it,
472 and that each has the expected contents, and that they are in the same
473 order as that in which they were appended.
475 self.assertEqual(len(mbox.listMessages()), 10)
477 [len(mbox.getMessage(i).read()) for i in range(10)],
482 class MaildirTestCase(unittest.TestCase):
484 self.d = self.mktemp()
485 mail.maildir.initializeMaildir(self.d)
488 shutil.rmtree(self.d)
490 def testInitializer(self):
492 trash = os.path.join(d, '.Trash')
494 self.failUnless(os.path.exists(d) and os.path.isdir(d))
495 self.failUnless(os.path.exists(os.path.join(d, 'new')))
496 self.failUnless(os.path.exists(os.path.join(d, 'cur')))
497 self.failUnless(os.path.exists(os.path.join(d, 'tmp')))
498 self.failUnless(os.path.isdir(os.path.join(d, 'new')))
499 self.failUnless(os.path.isdir(os.path.join(d, 'cur')))
500 self.failUnless(os.path.isdir(os.path.join(d, 'tmp')))
502 self.failUnless(os.path.exists(os.path.join(trash, 'new')))
503 self.failUnless(os.path.exists(os.path.join(trash, 'cur')))
504 self.failUnless(os.path.exists(os.path.join(trash, 'tmp')))
505 self.failUnless(os.path.isdir(os.path.join(trash, 'new')))
506 self.failUnless(os.path.isdir(os.path.join(trash, 'cur')))
507 self.failUnless(os.path.isdir(os.path.join(trash, 'tmp')))
510 def test_nameGenerator(self):
512 Each call to L{_MaildirNameGenerator.generate} returns a unique
513 string suitable for use as the basename of a new message file. The
514 names are ordered such that those generated earlier sort less than
515 those generated later.
519 generator = mail.maildir._MaildirNameGenerator(clock)
521 firstName = generator.generate()
523 secondName = generator.generate()
525 self.assertTrue(firstName < secondName)
528 def test_mailbox(self):
530 Exercise the methods of L{IMailbox} as implemented by
534 n = mail.maildir._generateMaildirName
535 msgs = [j(b, n()) for b in ('cur', 'new') for x in range(5)]
537 # Toss a few files into the mailbox
540 fObj = file(j(self.d, f), 'w')
545 mb = mail.maildir.MaildirMailbox(self.d)
546 self.assertEqual(mb.listMessages(), range(1, 11))
547 self.assertEqual(mb.listMessages(1), 2)
548 self.assertEqual(mb.listMessages(5), 6)
550 self.assertEqual(mb.getMessage(6).read(), 'x' * 7)
551 self.assertEqual(mb.getMessage(1).read(), 'x' * 2)
559 p, f = os.path.split(msgs[5])
562 self.assertEqual(mb.listMessages(5), 0)
563 self.failUnless(os.path.exists(j(self.d, '.Trash', 'cur', f)))
564 self.failIf(os.path.exists(j(self.d, msgs[5])))
566 mb.undeleteMessages()
567 self.assertEqual(mb.listMessages(5), 6)
568 self.failIf(os.path.exists(j(self.d, '.Trash', 'cur', f)))
569 self.failUnless(os.path.exists(j(self.d, msgs[5])))
571 class MaildirDirdbmDomainTestCase(unittest.TestCase):
573 self.P = self.mktemp()
574 self.S = mail.mail.MailService()
575 self.D = mail.maildir.MaildirDirdbmDomain(self.S, self.P)
578 shutil.rmtree(self.P)
580 def testAddUser(self):
581 toAdd = (('user1', 'pwd1'), ('user2', 'pwd2'), ('user3', 'pwd3'))
586 self.failUnless(u in self.D.dbm)
587 self.assertEqual(self.D.dbm[u], p)
588 self.failUnless(os.path.exists(os.path.join(self.P, u)))
590 def testCredentials(self):
591 creds = self.D.getCredentialsCheckers()
593 self.assertEqual(len(creds), 1)
594 self.failUnless(cred.checkers.ICredentialsChecker.providedBy(creds[0]))
595 self.failUnless(cred.credentials.IUsernamePassword in creds[0].credentialInterfaces)
597 def testRequestAvatar(self):
598 class ISomething(Interface):
601 self.D.addUser('user', 'password')
604 self.D.requestAvatar, 'user', None, ISomething
607 t = self.D.requestAvatar('user', None, pop3.IMailbox)
608 self.assertEqual(len(t), 3)
609 self.failUnless(t[0] is pop3.IMailbox)
610 self.failUnless(pop3.IMailbox.providedBy(t[1]))
614 def testRequestAvatarId(self):
615 self.D.addUser('user', 'password')
616 database = self.D.getCredentialsCheckers()[0]
618 creds = cred.credentials.UsernamePassword('user', 'wrong password')
620 cred.error.UnauthorizedLogin,
621 database.requestAvatarId, creds
624 creds = cred.credentials.UsernamePassword('user', 'password')
625 self.assertEqual(database.requestAvatarId(creds), 'user')
628 class StubAliasableDomain(object):
630 Minimal testable implementation of IAliasableDomain.
632 implements(mail.mail.IAliasableDomain)
634 def exists(self, user):
636 No test coverage for invocations of this method on domain objects,
637 so we just won't implement it.
639 raise NotImplementedError()
642 def addUser(self, user, password):
644 No test coverage for invocations of this method on domain objects,
645 so we just won't implement it.
647 raise NotImplementedError()
650 def getCredentialsCheckers(self):
652 This needs to succeed in order for other tests to complete
653 successfully, but we don't actually assert anything about its
654 behavior. Return an empty list. Sometime later we should return
655 something else and assert that a portal got set up properly.
660 def setAliasGroup(self, aliases):
662 Just record the value so the test can check it later.
664 self.aliasGroup = aliases
667 class ServiceDomainTestCase(unittest.TestCase):
669 self.S = mail.mail.MailService()
670 self.D = mail.protocols.DomainDeliveryBase(self.S, None)
671 self.D.service = self.S
672 self.D.protocolName = 'TEST'
673 self.D.host = 'hostname'
675 self.tmpdir = self.mktemp()
676 domain = mail.maildir.MaildirDirdbmDomain(self.S, self.tmpdir)
677 domain.addUser('user', 'password')
678 self.S.addDomain('test.domain', domain)
681 shutil.rmtree(self.tmpdir)
684 def testAddAliasableDomain(self):
686 Test that adding an IAliasableDomain to a mail service properly sets
687 up alias group references and such.
690 domain = StubAliasableDomain()
691 self.S.aliases = aliases
692 self.S.addDomain('example.com', domain)
693 self.assertIdentical(domain.aliasGroup, aliases)
696 def testReceivedHeader(self):
697 hdr = self.D.receivedHeader(
698 ('remotehost', '123.232.101.234'),
699 smtp.Address('<someguy@somplace>'),
702 fp = StringIO.StringIO(hdr)
703 m = rfc822.Message(fp)
704 self.assertEqual(len(m.items()), 1)
705 self.failUnless(m.has_key('Received'))
707 def testValidateTo(self):
708 user = smtp.User('user@test.domain', 'helo', None, 'wherever@whatever')
709 return defer.maybeDeferred(self.D.validateTo, user
710 ).addCallback(self._cbValidateTo
713 def _cbValidateTo(self, result):
714 self.failUnless(callable(result))
716 def testValidateToBadUsername(self):
717 user = smtp.User('resu@test.domain', 'helo', None, 'wherever@whatever')
718 return self.assertFailure(
719 defer.maybeDeferred(self.D.validateTo, user),
722 def testValidateToBadDomain(self):
723 user = smtp.User('user@domain.test', 'helo', None, 'wherever@whatever')
724 return self.assertFailure(
725 defer.maybeDeferred(self.D.validateTo, user),
728 def testValidateFrom(self):
729 helo = ('hostname', '127.0.0.1')
730 origin = smtp.Address('<user@hostname>')
731 self.failUnless(self.D.validateFrom(helo, origin) is origin)
733 helo = ('hostname', '1.2.3.4')
734 origin = smtp.Address('<user@hostname>')
735 self.failUnless(self.D.validateFrom(helo, origin) is origin)
737 helo = ('hostname', '1.2.3.4')
738 origin = smtp.Address('<>')
739 self.failUnless(self.D.validateFrom(helo, origin) is origin)
743 self.D.validateFrom, None, origin
746 class VirtualPOP3TestCase(unittest.TestCase):
748 self.tmpdir = self.mktemp()
749 self.S = mail.mail.MailService()
750 self.D = mail.maildir.MaildirDirdbmDomain(self.S, self.tmpdir)
751 self.D.addUser('user', 'password')
752 self.S.addDomain('test.domain', self.D)
754 portal = cred.portal.Portal(self.D)
755 map(portal.registerChecker, self.D.getCredentialsCheckers())
756 self.S.portals[''] = self.S.portals['test.domain'] = portal
758 self.P = mail.protocols.VirtualPOP3()
759 self.P.service = self.S
760 self.P.magic = '<unit test magic>'
763 shutil.rmtree(self.tmpdir)
765 def testAuthenticateAPOP(self):
766 resp = md5(self.P.magic + 'password').hexdigest()
767 return self.P.authenticateUserAPOP('user', resp
768 ).addCallback(self._cbAuthenticateAPOP
771 def _cbAuthenticateAPOP(self, result):
772 self.assertEqual(len(result), 3)
773 self.assertEqual(result[0], pop3.IMailbox)
774 self.failUnless(pop3.IMailbox.providedBy(result[1]))
777 def testAuthenticateIncorrectUserAPOP(self):
778 resp = md5(self.P.magic + 'password').hexdigest()
779 return self.assertFailure(
780 self.P.authenticateUserAPOP('resu', resp),
781 cred.error.UnauthorizedLogin)
783 def testAuthenticateIncorrectResponseAPOP(self):
784 resp = md5('wrong digest').hexdigest()
785 return self.assertFailure(
786 self.P.authenticateUserAPOP('user', resp),
787 cred.error.UnauthorizedLogin)
789 def testAuthenticatePASS(self):
790 return self.P.authenticateUserPASS('user', 'password'
791 ).addCallback(self._cbAuthenticatePASS
794 def _cbAuthenticatePASS(self, result):
795 self.assertEqual(len(result), 3)
796 self.assertEqual(result[0], pop3.IMailbox)
797 self.failUnless(pop3.IMailbox.providedBy(result[1]))
800 def testAuthenticateBadUserPASS(self):
801 return self.assertFailure(
802 self.P.authenticateUserPASS('resu', 'password'),
803 cred.error.UnauthorizedLogin)
805 def testAuthenticateBadPasswordPASS(self):
806 return self.assertFailure(
807 self.P.authenticateUserPASS('user', 'wrong password'),
808 cred.error.UnauthorizedLogin)
810 class empty(smtp.User):
814 class RelayTestCase(unittest.TestCase):
815 def testExists(self):
816 service = mail.mail.MailService()
817 domain = mail.relay.DomainQueuer(service)
820 address.UNIXAddress('/var/run/mail-relay'),
821 address.IPv4Address('TCP', '127.0.0.1', 12345),
825 address.IPv4Address('TCP', '192.168.2.1', 62),
826 address.IPv4Address('TCP', '1.2.3.4', 1943),
831 user.orig = 'user@host'
832 user.dest = 'tsoh@resu'
833 user.protocol = empty()
834 user.protocol.transport = empty()
835 user.protocol.transport.getPeer = lambda: peer
837 self.failUnless(callable(domain.exists(user)))
839 for peer in dontRelay:
841 user.orig = 'some@place'
842 user.protocol = empty()
843 user.protocol.transport = empty()
844 user.protocol.transport.getPeer = lambda: peer
845 user.dest = 'who@cares'
847 self.assertRaises(smtp.SMTPBadRcpt, domain.exists, user)
849 class RelayerTestCase(unittest.TestCase):
851 self.tmpdir = self.mktemp()
852 os.mkdir(self.tmpdir)
853 self.messageFiles = []
855 name = os.path.join(self.tmpdir, 'body-%d' % (i,))
856 f = file(name + '-H', 'w')
857 pickle.dump(['from-%d' % (i,), 'to-%d' % (i,)], f)
860 f = file(name + '-D', 'w')
863 self.messageFiles.append(name)
865 self.R = mail.relay.RelayerMixin()
866 self.R.loadMessages(self.messageFiles)
869 shutil.rmtree(self.tmpdir)
871 def testMailFrom(self):
873 self.assertEqual(self.R.getMailFrom(), 'from-%d' % (i,))
874 self.R.sentMail(250, None, None, None, None)
875 self.assertEqual(self.R.getMailFrom(), None)
877 def testMailTo(self):
879 self.assertEqual(self.R.getMailTo(), ['to-%d' % (i,)])
880 self.R.sentMail(250, None, None, None, None)
881 self.assertEqual(self.R.getMailTo(), None)
883 def testMailData(self):
885 name = os.path.join(self.tmpdir, 'body-%d' % (i,))
886 self.assertEqual(self.R.getMailData().read(), name)
887 self.R.sentMail(250, None, None, None, None)
888 self.assertEqual(self.R.getMailData(), None)
896 def notifySuccess(self, factory, message):
897 self.success.append((factory, message))
899 def notifyFailure(self, factory, message):
900 self.failure.append((factory, message))
902 def notifyDone(self, factory):
903 self.done.append(factory)
905 class ManagedRelayerTestCase(unittest.TestCase):
907 self.manager = Manager()
908 self.messages = range(0, 20, 2)
909 self.factory = object()
910 self.relay = mail.relaymanager.ManagedRelayerMixin(self.manager)
911 self.relay.messages = self.messages[:]
912 self.relay.names = self.messages[:]
913 self.relay.factory = self.factory
915 def testSuccessfulSentMail(self):
916 for i in self.messages:
917 self.relay.sentMail(250, None, None, None, None)
920 self.manager.success,
921 [(self.factory, m) for m in self.messages]
924 def testFailedSentMail(self):
925 for i in self.messages:
926 self.relay.sentMail(550, None, None, None, None)
929 self.manager.failure,
930 [(self.factory, m) for m in self.messages]
933 def testConnectionLost(self):
934 self.relay.connectionLost(failure.Failure(Exception()))
935 self.assertEqual(self.manager.done, [self.factory])
937 class DirectoryQueueTestCase(unittest.TestCase):
939 # This is almost a test case itself.
940 self.tmpdir = self.mktemp()
941 os.mkdir(self.tmpdir)
942 self.queue = mail.relaymanager.Queue(self.tmpdir)
943 self.queue.noisy = False
945 hdrF, msgF = self.queue.createNewMessage()
946 pickle.dump(['header', m], hdrF)
948 msgF.lineReceived('body: %d' % (m,))
950 self.queue.readDirectory()
953 shutil.rmtree(self.tmpdir)
955 def testWaiting(self):
956 self.failUnless(self.queue.hasWaiting())
957 self.assertEqual(len(self.queue.getWaiting()), 25)
959 waiting = self.queue.getWaiting()
960 self.queue.setRelaying(waiting[0])
961 self.assertEqual(len(self.queue.getWaiting()), 24)
963 self.queue.setWaiting(waiting[0])
964 self.assertEqual(len(self.queue.getWaiting()), 25)
966 def testRelaying(self):
967 for m in self.queue.getWaiting():
968 self.queue.setRelaying(m)
970 len(self.queue.getRelayed()),
971 25 - len(self.queue.getWaiting())
974 self.failIf(self.queue.hasWaiting())
976 relayed = self.queue.getRelayed()
977 self.queue.setWaiting(relayed[0])
978 self.assertEqual(len(self.queue.getWaiting()), 1)
979 self.assertEqual(len(self.queue.getRelayed()), 24)
982 msg = self.queue.getWaiting()[0]
983 self.queue.setRelaying(msg)
986 self.assertEqual(len(self.queue.getWaiting()), 24)
987 self.assertEqual(len(self.queue.getRelayed()), 0)
989 self.failIf(msg in self.queue.getWaiting())
990 self.failIf(msg in self.queue.getRelayed())
992 def testEnvelope(self):
995 for msg in self.queue.getWaiting():
996 envelopes.append(self.queue.getEnvelope(msg))
1005 from twisted.names import server
1006 from twisted.names import client
1007 from twisted.names import common
1009 class TestAuthority(common.ResolverBase):
1011 common.ResolverBase.__init__(self)
1014 def _lookup(self, name, cls, type, timeout = None):
1015 if name in self.addresses and type == dns.MX:
1017 for a in self.addresses[name]:
1019 name, dns.MX, dns.IN, 60, dns.Record_MX(0, a)
1022 return defer.succeed((results, [], []))
1023 return defer.fail(failure.Failure(dns.DomainError(name)))
1026 self.auth = TestAuthority()
1027 factory = server.DNSServerFactory([self.auth])
1028 protocol = dns.DNSDatagramProtocol(factory)
1030 self.port = reactor.listenTCP(0, factory, interface='127.0.0.1')
1031 portNumber = self.port.getHost().port
1034 self.udpPort = reactor.listenUDP(portNumber, protocol, interface='127.0.0.1')
1035 except CannotListenError:
1036 self.port.stopListening()
1039 self.resolver = client.Resolver(servers=[('127.0.0.1', portNumber)])
1042 def tearDownDNS(self):
1044 dl.append(defer.maybeDeferred(self.port.stopListening))
1045 dl.append(defer.maybeDeferred(self.udpPort.stopListening))
1046 if self.resolver.protocol.transport is not None:
1047 dl.append(defer.maybeDeferred(self.resolver.protocol.transport.stopListening))
1049 self.resolver._parseCall.cancel()
1052 return defer.DeferredList(dl)
1054 class MXTestCase(unittest.TestCase):
1056 Tests for L{mail.relaymanager.MXCalculator}.
1060 self.clock = task.Clock()
1061 self.mx = mail.relaymanager.MXCalculator(self.resolver, self.clock)
1064 return tearDownDNS(self)
1067 def test_defaultClock(self):
1069 L{MXCalculator}'s default clock is C{twisted.internet.reactor}.
1071 self.assertIdentical(
1072 mail.relaymanager.MXCalculator(self.resolver).clock,
1076 def testSimpleSuccess(self):
1077 self.auth.addresses['test.domain'] = ['the.email.test.domain']
1078 return self.mx.getMX('test.domain').addCallback(self._cbSimpleSuccess)
1080 def _cbSimpleSuccess(self, mx):
1081 self.assertEqual(mx.preference, 0)
1082 self.assertEqual(str(mx.name), 'the.email.test.domain')
1084 def testSimpleFailure(self):
1085 self.mx.fallbackToDomain = False
1086 return self.assertFailure(self.mx.getMX('test.domain'), IOError)
1088 def testSimpleFailureWithFallback(self):
1089 return self.assertFailure(self.mx.getMX('test.domain'), DNSLookupError)
1092 def _exchangeTest(self, domain, records, correctMailExchange):
1094 Issue an MX request for the given domain and arrange for it to be
1095 responded to with the given records. Verify that the resulting mail
1096 exchange is the indicated host.
1098 @type domain: C{str}
1099 @type records: C{list} of L{RRHeader}
1100 @type correctMailExchange: C{str}
1103 class DummyResolver(object):
1104 def lookupMailExchange(self, name):
1106 return defer.succeed((
1110 return defer.fail(DNSNameError(domain))
1112 self.mx.resolver = DummyResolver()
1113 d = self.mx.getMX(domain)
1114 def gotMailExchange(record):
1115 self.assertEqual(str(record.name), correctMailExchange)
1116 d.addCallback(gotMailExchange)
1120 def test_mailExchangePreference(self):
1122 The MX record with the lowest preference is returned by
1123 L{MXCalculator.getMX}.
1125 domain = "example.com"
1126 good = "good.example.com"
1127 bad = "bad.example.com"
1130 RRHeader(name=domain,
1131 type=Record_MX.TYPE,
1132 payload=Record_MX(1, bad)),
1133 RRHeader(name=domain,
1134 type=Record_MX.TYPE,
1135 payload=Record_MX(0, good)),
1136 RRHeader(name=domain,
1137 type=Record_MX.TYPE,
1138 payload=Record_MX(2, bad))]
1139 return self._exchangeTest(domain, records, good)
1142 def test_badExchangeExcluded(self):
1144 L{MXCalculator.getMX} returns the MX record with the lowest preference
1145 which is not also marked as bad.
1147 domain = "example.com"
1148 good = "good.example.com"
1149 bad = "bad.example.com"
1152 RRHeader(name=domain,
1153 type=Record_MX.TYPE,
1154 payload=Record_MX(0, bad)),
1155 RRHeader(name=domain,
1156 type=Record_MX.TYPE,
1157 payload=Record_MX(1, good))]
1158 self.mx.markBad(bad)
1159 return self._exchangeTest(domain, records, good)
1162 def test_fallbackForAllBadExchanges(self):
1164 L{MXCalculator.getMX} returns the MX record with the lowest preference
1165 if all the MX records in the response have been marked bad.
1167 domain = "example.com"
1168 bad = "bad.example.com"
1169 worse = "worse.example.com"
1172 RRHeader(name=domain,
1173 type=Record_MX.TYPE,
1174 payload=Record_MX(0, bad)),
1175 RRHeader(name=domain,
1176 type=Record_MX.TYPE,
1177 payload=Record_MX(1, worse))]
1178 self.mx.markBad(bad)
1179 self.mx.markBad(worse)
1180 return self._exchangeTest(domain, records, bad)
1183 def test_badExchangeExpires(self):
1185 L{MXCalculator.getMX} returns the MX record with the lowest preference
1186 if it was last marked bad longer than L{MXCalculator.timeOutBadMX}
1189 domain = "example.com"
1190 good = "good.example.com"
1191 previouslyBad = "bad.example.com"
1194 RRHeader(name=domain,
1195 type=Record_MX.TYPE,
1196 payload=Record_MX(0, previouslyBad)),
1197 RRHeader(name=domain,
1198 type=Record_MX.TYPE,
1199 payload=Record_MX(1, good))]
1200 self.mx.markBad(previouslyBad)
1201 self.clock.advance(self.mx.timeOutBadMX)
1202 return self._exchangeTest(domain, records, previouslyBad)
1205 def test_goodExchangeUsed(self):
1207 L{MXCalculator.getMX} returns the MX record with the lowest preference
1208 if it was marked good after it was marked bad.
1210 domain = "example.com"
1211 good = "good.example.com"
1212 previouslyBad = "bad.example.com"
1215 RRHeader(name=domain,
1216 type=Record_MX.TYPE,
1217 payload=Record_MX(0, previouslyBad)),
1218 RRHeader(name=domain,
1219 type=Record_MX.TYPE,
1220 payload=Record_MX(1, good))]
1221 self.mx.markBad(previouslyBad)
1222 self.mx.markGood(previouslyBad)
1223 self.clock.advance(self.mx.timeOutBadMX)
1224 return self._exchangeTest(domain, records, previouslyBad)
1227 def test_successWithoutResults(self):
1229 If an MX lookup succeeds but the result set is empty,
1230 L{MXCalculator.getMX} should try to look up an I{A} record for the
1231 requested name and call back its returned Deferred with that
1235 domain = 'example.org'
1237 class DummyResolver(object):
1239 Fake resolver which will respond to an MX lookup with an empty
1242 @ivar mx: A dictionary mapping hostnames to three-tuples of
1243 results to be returned from I{MX} lookups.
1245 @ivar a: A dictionary mapping hostnames to addresses to be
1246 returned from I{A} lookups.
1248 mx = {domain: ([], [], [])}
1251 def lookupMailExchange(self, domain):
1252 return defer.succeed(self.mx[domain])
1254 def getHostByName(self, domain):
1255 return defer.succeed(self.a[domain])
1257 self.mx.resolver = DummyResolver()
1258 d = self.mx.getMX(domain)
1259 d.addCallback(self.assertEqual, Record_MX(name=ip))
1263 def test_failureWithSuccessfulFallback(self):
1265 Test that if the MX record lookup fails, fallback is enabled, and an A
1266 record is available for the name, then the Deferred returned by
1267 L{MXCalculator.getMX} ultimately fires with a Record_MX instance which
1268 gives the address in the A record for the name.
1270 class DummyResolver(object):
1272 Fake resolver which will fail an MX lookup but then succeed a
1275 def lookupMailExchange(self, domain):
1276 return defer.fail(DNSNameError())
1278 def getHostByName(self, domain):
1279 return defer.succeed("1.2.3.4")
1281 self.mx.resolver = DummyResolver()
1282 d = self.mx.getMX("domain")
1283 d.addCallback(self.assertEqual, Record_MX(name="1.2.3.4"))
1287 def test_cnameWithoutGlueRecords(self):
1289 If an MX lookup returns a single CNAME record as a result, MXCalculator
1290 will perform an MX lookup for the canonical name indicated and return
1291 the MX record which results.
1293 alias = "alias.example.com"
1294 canonical = "canonical.example.com"
1295 exchange = "mail.example.com"
1297 class DummyResolver(object):
1299 Fake resolver which will return a CNAME for an MX lookup of a name
1300 which is an alias and an MX for an MX lookup of the canonical name.
1302 def lookupMailExchange(self, domain):
1304 return defer.succeed((
1305 [RRHeader(name=domain,
1306 type=Record_CNAME.TYPE,
1307 payload=Record_CNAME(canonical))],
1309 elif domain == canonical:
1310 return defer.succeed((
1311 [RRHeader(name=domain,
1312 type=Record_MX.TYPE,
1313 payload=Record_MX(0, exchange))],
1316 return defer.fail(DNSNameError(domain))
1318 self.mx.resolver = DummyResolver()
1319 d = self.mx.getMX(alias)
1320 d.addCallback(self.assertEqual, Record_MX(name=exchange))
1324 def test_cnameChain(self):
1326 If L{MXCalculator.getMX} encounters a CNAME chain which is longer than
1327 the length specified, the returned L{Deferred} should errback with
1328 L{CanonicalNameChainTooLong}.
1330 class DummyResolver(object):
1332 Fake resolver which generates a CNAME chain of infinite length in
1333 response to MX lookups.
1337 def lookupMailExchange(self, domain):
1338 self.chainCounter += 1
1339 name = 'x-%d.example.com' % (self.chainCounter,)
1340 return defer.succeed((
1341 [RRHeader(name=domain,
1342 type=Record_CNAME.TYPE,
1343 payload=Record_CNAME(name))],
1347 self.mx.resolver = DummyResolver()
1348 d = self.mx.getMX("mail.example.com", cnameLimit)
1350 d, twisted.mail.relaymanager.CanonicalNameChainTooLong)
1351 def cbChainTooLong(error):
1352 self.assertEqual(error.args[0], Record_CNAME("x-%d.example.com" % (cnameLimit + 1,)))
1353 self.assertEqual(self.mx.resolver.chainCounter, cnameLimit + 1)
1354 d.addCallback(cbChainTooLong)
1358 def test_cnameWithGlueRecords(self):
1360 If an MX lookup returns a CNAME and the MX record for the CNAME, the
1361 L{Deferred} returned by L{MXCalculator.getMX} should be called back
1362 with the name from the MX record without further lookups being
1366 alias = "alias.example.com"
1367 canonical = "canonical.example.com"
1368 exchange = "mail.example.com"
1370 class DummyResolver(object):
1371 def lookupMailExchange(self, domain):
1372 if domain != alias or lookedUp:
1373 # Don't give back any results for anything except the alias
1374 # or on any request after the first.
1376 return defer.succeed((
1377 [RRHeader(name=alias,
1378 type=Record_CNAME.TYPE,
1379 payload=Record_CNAME(canonical)),
1380 RRHeader(name=canonical,
1381 type=Record_MX.TYPE,
1382 payload=Record_MX(name=exchange))],
1385 self.mx.resolver = DummyResolver()
1386 d = self.mx.getMX(alias)
1387 d.addCallback(self.assertEqual, Record_MX(name=exchange))
1391 def test_cnameLoopWithGlueRecords(self):
1393 If an MX lookup returns two CNAME records which point to each other,
1394 the loop should be detected and the L{Deferred} returned by
1395 L{MXCalculator.getMX} should be errbacked with L{CanonicalNameLoop}.
1397 firstAlias = "cname1.example.com"
1398 secondAlias = "cname2.example.com"
1400 class DummyResolver(object):
1401 def lookupMailExchange(self, domain):
1402 return defer.succeed((
1403 [RRHeader(name=firstAlias,
1404 type=Record_CNAME.TYPE,
1405 payload=Record_CNAME(secondAlias)),
1406 RRHeader(name=secondAlias,
1407 type=Record_CNAME.TYPE,
1408 payload=Record_CNAME(firstAlias))],
1411 self.mx.resolver = DummyResolver()
1412 d = self.mx.getMX(firstAlias)
1413 self.assertFailure(d, twisted.mail.relaymanager.CanonicalNameLoop)
1417 def testManyRecords(self):
1418 self.auth.addresses['test.domain'] = [
1419 'mx1.test.domain', 'mx2.test.domain', 'mx3.test.domain'
1421 return self.mx.getMX('test.domain'
1422 ).addCallback(self._cbManyRecordsSuccessfulLookup
1425 def _cbManyRecordsSuccessfulLookup(self, mx):
1426 self.failUnless(str(mx.name).split('.', 1)[0] in ('mx1', 'mx2', 'mx3'))
1427 self.mx.markBad(str(mx.name))
1428 return self.mx.getMX('test.domain'
1429 ).addCallback(self._cbManyRecordsDifferentResult, mx
1432 def _cbManyRecordsDifferentResult(self, nextMX, mx):
1433 self.assertNotEqual(str(mx.name), str(nextMX.name))
1434 self.mx.markBad(str(nextMX.name))
1436 return self.mx.getMX('test.domain'
1437 ).addCallback(self._cbManyRecordsLastResult, mx, nextMX
1440 def _cbManyRecordsLastResult(self, lastMX, mx, nextMX):
1441 self.assertNotEqual(str(mx.name), str(lastMX.name))
1442 self.assertNotEqual(str(nextMX.name), str(lastMX.name))
1444 self.mx.markBad(str(lastMX.name))
1445 self.mx.markGood(str(nextMX.name))
1447 return self.mx.getMX('test.domain'
1448 ).addCallback(self._cbManyRecordsRepeatSpecificResult, nextMX
1451 def _cbManyRecordsRepeatSpecificResult(self, againMX, nextMX):
1452 self.assertEqual(str(againMX.name), str(nextMX.name))
1454 class LiveFireExercise(unittest.TestCase):
1455 if interfaces.IReactorUDP(reactor, None) is None:
1456 skip = "UDP support is required to determining MX records"
1461 'domainDir', 'insertionDomain', 'insertionQueue',
1462 'destinationDomain', 'destinationQueue'
1466 for d in self.tmpdirs:
1467 if os.path.exists(d):
1469 return tearDownDNS(self)
1471 def testLocalDelivery(self):
1472 service = mail.mail.MailService()
1473 service.smtpPortal.registerChecker(cred.checkers.AllowAnonymousAccess())
1474 domain = mail.maildir.MaildirDirdbmDomain(service, 'domainDir')
1475 domain.addUser('user', 'password')
1476 service.addDomain('test.domain', domain)
1477 service.portals[''] = service.portals['test.domain']
1478 map(service.portals[''].registerChecker, domain.getCredentialsCheckers())
1480 service.setQueue(mail.relay.DomainQueuer(service))
1481 manager = mail.relaymanager.SmartHostSMTPRelayingManager(service.queue, None)
1482 helper = mail.relaymanager.RelayStateHelper(manager, 1)
1484 f = service.getSMTPFactory()
1486 self.smtpServer = reactor.listenTCP(0, f, interface='127.0.0.1')
1488 client = LineSendingProtocol([
1490 'MAIL FROM: <user@hostname>',
1491 'RCPT TO: <user@test.domain>',
1493 'This is the message',
1499 f = protocol.ClientFactory()
1500 f.protocol = lambda: client
1501 f.clientConnectionLost = lambda *args: done.callback(None)
1502 reactor.connectTCP('127.0.0.1', self.smtpServer.getHost().port, f)
1505 mbox = domain.requestAvatar('user', None, pop3.IMailbox)[1]
1506 msg = mbox.getMessage(0).read()
1507 self.failIfEqual(msg.find('This is the message'), -1)
1509 return self.smtpServer.stopListening()
1510 done.addCallback(finished)
1514 def testRelayDelivery(self):
1515 # Here is the service we will connect to and send mail from
1516 insServ = mail.mail.MailService()
1517 insServ.smtpPortal.registerChecker(cred.checkers.AllowAnonymousAccess())
1518 domain = mail.maildir.MaildirDirdbmDomain(insServ, 'insertionDomain')
1519 insServ.addDomain('insertion.domain', domain)
1520 os.mkdir('insertionQueue')
1521 insServ.setQueue(mail.relaymanager.Queue('insertionQueue'))
1522 insServ.domains.setDefaultDomain(mail.relay.DomainQueuer(insServ))
1523 manager = mail.relaymanager.SmartHostSMTPRelayingManager(insServ.queue)
1524 manager.fArgs += ('test.identity.hostname',)
1525 helper = mail.relaymanager.RelayStateHelper(manager, 1)
1526 # Yoink! Now the internet obeys OUR every whim!
1527 manager.mxcalc = mail.relaymanager.MXCalculator(self.resolver)
1528 # And this is our whim.
1529 self.auth.addresses['destination.domain'] = ['127.0.0.1']
1531 f = insServ.getSMTPFactory()
1532 self.insServer = reactor.listenTCP(0, f, interface='127.0.0.1')
1534 # Here is the service the previous one will connect to for final
1536 destServ = mail.mail.MailService()
1537 destServ.smtpPortal.registerChecker(cred.checkers.AllowAnonymousAccess())
1538 domain = mail.maildir.MaildirDirdbmDomain(destServ, 'destinationDomain')
1539 domain.addUser('user', 'password')
1540 destServ.addDomain('destination.domain', domain)
1541 os.mkdir('destinationQueue')
1542 destServ.setQueue(mail.relaymanager.Queue('destinationQueue'))
1543 manager2 = mail.relaymanager.SmartHostSMTPRelayingManager(destServ.queue)
1544 helper = mail.relaymanager.RelayStateHelper(manager, 1)
1545 helper.startService()
1547 f = destServ.getSMTPFactory()
1548 self.destServer = reactor.listenTCP(0, f, interface='127.0.0.1')
1550 # Update the port number the *first* relay will connect to, because we can't use
1552 manager.PORT = self.destServer.getHost().port
1554 client = LineSendingProtocol([
1556 'MAIL FROM: <user@wherever>',
1557 'RCPT TO: <user@destination.domain>',
1559 'This is the message',
1565 f = protocol.ClientFactory()
1566 f.protocol = lambda: client
1567 f.clientConnectionLost = lambda *args: done.callback(None)
1568 reactor.connectTCP('127.0.0.1', self.insServer.getHost().port, f)
1571 # First part of the delivery is done. Poke the queue manually now
1572 # so we don't have to wait for the queue to be flushed.
1573 delivery = manager.checkState()
1575 mbox = domain.requestAvatar('user', None, pop3.IMailbox)[1]
1576 msg = mbox.getMessage(0).read()
1577 self.failIfEqual(msg.find('This is the message'), -1)
1579 self.insServer.stopListening()
1580 self.destServer.stopListening()
1581 helper.stopService()
1582 delivery.addCallback(delivered)
1584 done.addCallback(finished)
1588 aliasFile = StringIO.StringIO("""\
1591 testuser: address1,address2, address3,
1592 continuation@address, |/bin/process/this
1594 usertwo:thisaddress,thataddress, lastaddress
1595 lastuser: :/includable, /filename, |/program, address
1598 class LineBufferMessage:
1604 def lineReceived(self, line):
1605 self.lines.append(line)
1607 def eomReceived(self):
1609 return defer.succeed('<Whatever>')
1611 def connectionLost(self):
1614 class AliasTestCase(unittest.TestCase):
1619 'After a blank line',
1626 def testHandle(self):
1629 'user: another@host\n',
1630 'nextuser: |/bin/program\n',
1632 'moreusers: :/etc/include/filename\n',
1633 'multiuser: first@host, second@host,last@anotherhost',
1637 mail.alias.handle(result, l, 'TestCase', None)
1639 self.assertEqual(result['user'], ['another@host', 'me@again'])
1640 self.assertEqual(result['nextuser'], ['|/bin/program'])
1641 self.assertEqual(result['moreusers'], [':/etc/include/filename'])
1642 self.assertEqual(result['multiuser'], ['first@host', 'second@host', 'last@anotherhost'])
1644 def testFileLoader(self):
1645 domains = {'': object()}
1646 result = mail.alias.loadAliasFile(domains, fp=aliasFile)
1648 self.assertEqual(len(result), 3)
1650 group = result['testuser']
1652 for a in ('address1', 'address2', 'address3', 'continuation@address', '/bin/process/this'):
1653 self.failIfEqual(s.find(a), -1)
1654 self.assertEqual(len(group), 5)
1656 group = result['usertwo']
1658 for a in ('thisaddress', 'thataddress', 'lastaddress'):
1659 self.failIfEqual(s.find(a), -1)
1660 self.assertEqual(len(group), 3)
1662 group = result['lastuser']
1664 self.assertEqual(s.find('/includable'), -1)
1665 for a in ('/filename', 'program', 'address'):
1666 self.failIfEqual(s.find(a), -1, '%s not found' % a)
1667 self.assertEqual(len(group), 3)
1669 def testMultiWrapper(self):
1670 msgs = LineBufferMessage(), LineBufferMessage(), LineBufferMessage()
1671 msg = mail.alias.MultiWrapper(msgs)
1673 for L in self.lines:
1675 return msg.eomReceived().addCallback(self._cbMultiWrapper, msgs)
1677 def _cbMultiWrapper(self, ignored, msgs):
1679 self.failUnless(m.eom)
1681 self.assertEqual(self.lines, m.lines)
1683 def testFileAlias(self):
1684 tmpfile = self.mktemp()
1685 a = mail.alias.FileAlias(tmpfile, None, None)
1686 m = a.createMessageReceiver()
1688 for l in self.lines:
1690 return m.eomReceived().addCallback(self._cbTestFileAlias, tmpfile)
1692 def _cbTestFileAlias(self, ignored, tmpfile):
1693 lines = file(tmpfile).readlines()
1694 self.assertEqual([L[:-1] for L in lines], self.lines)
1698 class DummyProcess(object):
1699 __slots__ = ['onEnd']
1703 class MockProcessAlias(mail.alias.ProcessAlias):
1705 A alias processor that doesn't actually launch processes.
1708 def spawnProcess(self, proto, program, path):
1710 Don't spawn a process.
1715 class MockAliasGroup(mail.alias.AliasGroup):
1717 An alias group using C{MockProcessAlias}.
1719 processAliasFactory = MockProcessAlias
1723 class StubProcess(object):
1725 Fake implementation of L{IProcessTransport}.
1727 @ivar signals: A list of all the signals which have been sent to this fake
1734 def loseConnection(self):
1736 No-op implementation of disconnection.
1740 def signalProcess(self, signal):
1742 Record a signal sent to this process for later inspection.
1744 self.signals.append(signal)
1748 class ProcessAliasTestCase(unittest.TestCase):
1750 Tests for alias resolution.
1752 if interfaces.IReactorProcess(reactor, None) is None:
1753 skip = "IReactorProcess not supported"
1759 'After a blank line',
1763 def exitStatus(self, code):
1765 Construct a status from the given exit code.
1767 @type code: L{int} between 0 and 255 inclusive.
1768 @param code: The exit status which the code will represent.
1771 @return: A status integer for the given exit code.
1773 # /* Macros for constructing status values. */
1774 # #define __W_EXITCODE(ret, sig) ((ret) << 8 | (sig))
1775 status = (code << 8) | 0
1778 self.assertTrue(os.WIFEXITED(status))
1779 self.assertEqual(os.WEXITSTATUS(status), code)
1780 self.assertFalse(os.WIFSIGNALED(status))
1785 def signalStatus(self, signal):
1787 Construct a status from the given signal.
1789 @type signal: L{int} between 0 and 255 inclusive.
1790 @param signal: The signal number which the status will represent.
1793 @return: A status integer for the given signal.
1795 # /* If WIFSIGNALED(STATUS), the terminating signal. */
1796 # #define __WTERMSIG(status) ((status) & 0x7f)
1797 # /* Nonzero if STATUS indicates termination by a signal. */
1798 # #define __WIFSIGNALED(status) \
1799 # (((signed char) (((status) & 0x7f) + 1) >> 1) > 0)
1803 self.assertTrue(os.WIFSIGNALED(status))
1804 self.assertEqual(os.WTERMSIG(status), signal)
1805 self.assertFalse(os.WIFEXITED(status))
1812 Replace L{smtp.DNSNAME} with a well-known value.
1814 self.DNSNAME = smtp.DNSNAME
1820 Restore the original value of L{smtp.DNSNAME}.
1822 smtp.DNSNAME = self.DNSNAME
1825 def test_processAlias(self):
1827 Standard call to C{mail.alias.ProcessAlias}: check that the specified
1828 script is called, and that the input is correctly transferred to it.
1830 sh = FilePath(self.mktemp())
1833 rm -f process.alias.out
1835 echo $i >> process.alias.out
1837 os.chmod(sh.path, 0700)
1838 a = mail.alias.ProcessAlias(sh.path, None, None)
1839 m = a.createMessageReceiver()
1841 for l in self.lines:
1844 def _cbProcessAlias(ignored):
1845 lines = file('process.alias.out').readlines()
1846 self.assertEqual([L[:-1] for L in lines], self.lines)
1848 return m.eomReceived().addCallback(_cbProcessAlias)
1851 def test_processAliasTimeout(self):
1853 If the alias child process does not exit within a particular period of
1854 time, the L{Deferred} returned by L{MessageWrapper.eomReceived} should
1855 fail with L{ProcessAliasTimeout} and send the I{KILL} signal to the
1858 reactor = task.Clock()
1859 transport = StubProcess()
1860 proto = mail.alias.ProcessAliasProtocol()
1861 proto.makeConnection(transport)
1863 receiver = mail.alias.MessageWrapper(proto, None, reactor)
1864 d = receiver.eomReceived()
1865 reactor.advance(receiver.completionTimeout)
1866 def timedOut(ignored):
1867 self.assertEqual(transport.signals, ['KILL'])
1868 # Now that it has been killed, disconnect the protocol associated
1871 ProcessTerminated(self.signalStatus(signal.SIGKILL)))
1872 self.assertFailure(d, mail.alias.ProcessAliasTimeout)
1873 d.addCallback(timedOut)
1877 def test_earlyProcessTermination(self):
1879 If the process associated with an L{mail.alias.MessageWrapper} exits
1880 before I{eomReceived} is called, the L{Deferred} returned by
1881 I{eomReceived} should fail.
1883 transport = StubProcess()
1884 protocol = mail.alias.ProcessAliasProtocol()
1885 protocol.makeConnection(transport)
1886 receiver = mail.alias.MessageWrapper(protocol, None, None)
1887 protocol.processEnded(failure.Failure(ProcessDone(0)))
1888 return self.assertFailure(receiver.eomReceived(), ProcessDone)
1891 def _terminationTest(self, status):
1893 Verify that if the process associated with an
1894 L{mail.alias.MessageWrapper} exits with the given status, the
1895 L{Deferred} returned by I{eomReceived} fails with L{ProcessTerminated}.
1897 transport = StubProcess()
1898 protocol = mail.alias.ProcessAliasProtocol()
1899 protocol.makeConnection(transport)
1900 receiver = mail.alias.MessageWrapper(protocol, None, None)
1901 protocol.processEnded(
1902 failure.Failure(ProcessTerminated(status)))
1903 return self.assertFailure(receiver.eomReceived(), ProcessTerminated)
1906 def test_errorProcessTermination(self):
1908 If the process associated with an L{mail.alias.MessageWrapper} exits
1909 with a non-zero exit code, the L{Deferred} returned by I{eomReceived}
1912 return self._terminationTest(self.exitStatus(1))
1915 def test_signalProcessTermination(self):
1917 If the process associated with an L{mail.alias.MessageWrapper} exits
1918 because it received a signal, the L{Deferred} returned by
1919 I{eomReceived} should fail.
1921 return self._terminationTest(self.signalStatus(signal.SIGHUP))
1924 def test_aliasResolution(self):
1926 Check that the C{resolve} method of alias processors produce the correct
1928 - direct alias with L{mail.alias.AddressAlias} if a simple input is passed
1929 - aliases in a file with L{mail.alias.FileWrapper} if an input in the format
1931 - aliases resulting of a process call wrapped by L{mail.alias.MessageWrapper}
1932 if the format is '|process'
1935 domain = {'': TestDomain(aliases, ['user1', 'user2', 'user3'])}
1936 A1 = MockAliasGroup(['user1', '|echo', '/file'], domain, 'alias1')
1937 A2 = MockAliasGroup(['user2', 'user3'], domain, 'alias2')
1938 A3 = mail.alias.AddressAlias('alias1', domain, 'alias3')
1945 res1 = A1.resolve(aliases)
1946 r1 = map(str, res1.objs)
1948 expected = map(str, [
1949 mail.alias.AddressAlias('user1', None, None),
1950 mail.alias.MessageWrapper(DummyProcess(), 'echo'),
1951 mail.alias.FileWrapper('/file'),
1954 self.assertEqual(r1, expected)
1956 res2 = A2.resolve(aliases)
1957 r2 = map(str, res2.objs)
1959 expected = map(str, [
1960 mail.alias.AddressAlias('user2', None, None),
1961 mail.alias.AddressAlias('user3', None, None)
1964 self.assertEqual(r2, expected)
1966 res3 = A3.resolve(aliases)
1967 r3 = map(str, res3.objs)
1969 expected = map(str, [
1970 mail.alias.AddressAlias('user1', None, None),
1971 mail.alias.MessageWrapper(DummyProcess(), 'echo'),
1972 mail.alias.FileWrapper('/file'),
1975 self.assertEqual(r3, expected)
1978 def test_cyclicAlias(self):
1980 Check that a cycle in alias resolution is correctly handled.
1983 domain = {'': TestDomain(aliases, [])}
1984 A1 = mail.alias.AddressAlias('alias2', domain, 'alias1')
1985 A2 = mail.alias.AddressAlias('alias3', domain, 'alias2')
1986 A3 = mail.alias.AddressAlias('alias1', domain, 'alias3')
1993 self.assertEqual(aliases['alias1'].resolve(aliases), None)
1994 self.assertEqual(aliases['alias2'].resolve(aliases), None)
1995 self.assertEqual(aliases['alias3'].resolve(aliases), None)
1997 A4 = MockAliasGroup(['|echo', 'alias1'], domain, 'alias4')
1998 aliases['alias4'] = A4
2000 res = A4.resolve(aliases)
2001 r = map(str, res.objs)
2003 expected = map(str, [
2004 mail.alias.MessageWrapper(DummyProcess(), 'echo')
2007 self.assertEqual(r, expected)
2015 def __init__(self, aliases, users):
2016 self.aliases = aliases
2019 def exists(self, user, memo=None):
2020 user = user.dest.local
2021 if user in self.users:
2022 return lambda: mail.alias.AddressAlias(user, None, None)
2024 a = self.aliases[user]
2026 raise smtp.SMTPBadRcpt(user)
2028 aliases = a.resolve(self.aliases, memo)
2030 return lambda: aliases
2031 raise smtp.SMTPBadRcpt(user)
2034 from twisted.python.runtime import platformType
2036 if platformType != "posix":
2037 for o in locals().values():
2038 if isinstance(o, (types.ClassType, type)) and issubclass(o, unittest.TestCase):
2039 o.skip = "twisted.mail only works on posix"