[Zope3-checkins] CVS: Zope3/src/zope/app/mail/tests - test_directives.py:1.3.2.2 test_service.py:1.1.2.2
Albertas Agejevas
alga@codeworks.lt
Mon, 26 May 2003 14:25:50 -0400
Update of /cvs-repository/Zope3/src/zope/app/mail/tests
In directory cvs.zope.org:/tmp/cvs-serv4417/tests
Modified Files:
Tag: cw-mail-branch
test_directives.py test_service.py
Log Message:
ZCML directives for the mailers.
A thread for queued mail delivery.
=== Zope3/src/zope/app/mail/tests/test_directives.py 1.3.2.1 => 1.3.2.2 ===
--- Zope3/src/zope/app/mail/tests/test_directives.py:1.3.2.1 Thu May 22 04:51:41 2003
+++ Zope3/src/zope/app/mail/tests/test_directives.py Mon May 26 14:25:49 2003
@@ -16,6 +16,7 @@
$Id$
"""
import unittest
+import threading
from cStringIO import StringIO
@@ -36,6 +37,20 @@
</zopeConfigure>"""
+class MaildirStub:
+
+ def __init__(self, path, create=False):
+ self.path = path
+ self.create = create
+
+ def __iter__(self):
+ return iter(())
+
+ def newMessage(self):
+ return None
+
+
+
class DirectivesTest(PlacelessSetup, unittest.TestCase):
def setUp(self):
@@ -46,18 +61,29 @@
zope.app.interfaces.mail.IMailService)
XMLConfig('metameta.zcml', zope.configuration)()
XMLConfig('meta.zcml', zope.app.mail)()
+ from zope.app.mail import service
+ self.orig_maildir = service.Maildir
+ service.Maildir = MaildirStub
+
+ def tearDown(self):
+ from zope.app.mail import service
+ service.Maildir = self.orig_maildir
def testQueuedService(self):
+ threads = threading.activeCount()
+ provideMailer("smtp", object())
xmlconfig(StringIO(template % (
'''
<mail:queuedService name="Mail"
queuePath="/path/to/mailbox"
+ mailer="smtp"
permission="zope.Public" />
'''
)), None, Context([], zope.app.mail))
service = getService(None, 'Mail')
self.assertEqual('QueuedMailService', service.__class__.__name__)
self.assertEqual('/path/to/mailbox', service.queuePath)
+ self.assertEqual(threading.activeCount(), threads + 1)
def testDirectService(self):
testMailer = object()
@@ -73,6 +99,40 @@
self.assertEqual('DirectMailService', service.__class__.__name__)
self.assert_(testMailer is service.mailer)
+
+ def testSendmailMailer(self):
+ from zope.app.interfaces.mail import ISendmailMailer
+ from zope.app.mail.metaconfigure import queryMailer
+ xmlconfig(StringIO(template % (
+ '''
+ <mail:sendmailMailer name="Sendmail"
+ command="/usr/lib/sendmail -oem -oi -f %(from)s %(to)s" />
+ '''
+ )), None, Context([], zope.app.mail))
+ self.assert_(ISendmailMailer.isImplementedBy(queryMailer("Sendmail")))
+
+ def testSMTPMailer(self):
+ from zope.app.interfaces.mail import ISMTPMailer
+ from zope.app.mail.metaconfigure import queryMailer
+ xmlconfig(StringIO(template % (
+ '''
+ <mail:smtpMailer name="smtp"
+ hostname="localhost"
+ port="25"
+ username="zope3"
+ password="xyzzy"
+ />
+ '''
+ )), None, Context([], zope.app.mail))
+
+ xmlconfig(StringIO(template % (
+ '''
+ <mail:smtpMailer name="smtp2"
+ hostname="smarthost"
+ />
+ '''
+ )), None, Context([], zope.app.mail))
+ self.assert_(ISMTPMailer.isImplementedBy(queryMailer("smtp")))
def test_suite():
return unittest.makeSuite(DirectivesTest)
=== Zope3/src/zope/app/mail/tests/test_service.py 1.1.2.1 => 1.1.2.2 ===
--- Zope3/src/zope/app/mail/tests/test_service.py:1.1.2.1 Thu May 22 04:51:41 2003
+++ Zope3/src/zope/app/mail/tests/test_service.py Mon May 26 14:25:49 2003
@@ -22,6 +22,8 @@
from zope.interface.verify import verifyObject
from zope.app.interfaces.mail import IMailer
from transaction import get_transaction
+from os import mkdir, rmdir, unlink, getpid, listdir
+import os.path
__metaclass__ = type
@@ -29,11 +31,11 @@
class MailerStub:
implements(IMailer)
-
- sent_messages = ()
+ def __init__(self, *args, **kw):
+ self.sent_messages = []
def send(self, fromaddr, toaddrs, message):
- self.sent_messages += ((fromaddr, toaddrs, message), )
+ self.sent_messages.append((fromaddr, toaddrs, message))
class TestMailDataManager(TestCase):
@@ -74,15 +76,15 @@
msgid = service.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(msgid, '20030519.1234@example.org')
- self.assertEquals(mailer.sent_messages, ())
+ self.assertEquals(mailer.sent_messages, [])
get_transaction().commit()
self.assertEquals(mailer.sent_messages,
- ((fromaddr, toaddrs, opt_headers + message), ))
+ [(fromaddr, toaddrs, opt_headers + message)])
- mailer.sent_messages = ()
+ mailer.sent_messages = []
msgid = service.send(fromaddr, toaddrs, message)
self.assert_('@' in msgid)
- self.assertEquals(mailer.sent_messages, ())
+ self.assertEquals(mailer.sent_messages, [])
get_transaction().commit()
self.assertEquals(len(mailer.sent_messages), 1)
self.assertEquals(mailer.sent_messages[0][0], fromaddr)
@@ -91,11 +93,11 @@
new_headers = mailer.sent_messages[0][2][:-len(message)]
self.assert_(new_headers.find('Message-Id: <%s>' % msgid) != -1)
- mailer.sent_messages = ()
+ mailer.sent_messages = []
msgid = service.send(fromaddr, toaddrs, opt_headers + message)
- self.assertEquals(mailer.sent_messages, ())
+ self.assertEquals(mailer.sent_messages, [])
get_transaction().abort()
- self.assertEquals(mailer.sent_messages, ())
+ self.assertEquals(mailer.sent_messages, [])
class MaildirWriterStub:
@@ -124,12 +126,15 @@
def __init__(self, path, create=False):
self.path = path
self.create = create
+ self.msgs = []
def __iter__(self):
- return iter(())
+ return iter(self.msgs)
def newMessage(self):
- return MaildirWriterStub()
+ m = MaildirWriterStub()
+ self.msgs.append(m)
+ return m
class TestQueuedMailService(TestCase):
@@ -153,9 +158,11 @@
def testSend(self):
from zope.app.mail.service import QueuedMailService
service = QueuedMailService('/path/to/mailbox')
- fromaddr = 'Jim <jim@example.com'
- toaddrs = ('Guido <guido@example.com>',
- 'Steve <steve@examplecom>')
+ fromaddr = 'jim@example.com'
+ toaddrs = ('guido@example.com',
+ 'steve@examplecom')
+ zope_headers = ('X-Zope-From: jim@example.com\n'
+ 'X-Zope-To: guido@example.com, steve@examplecom\n')
opt_headers = ('From: Jim <jim@example.org>\n'
'To: some-zope-coders:;\n'
'Date: Mon, 19 May 2003 10:17:36 -0400\n'
@@ -170,7 +177,7 @@
self.assertEquals(MaildirWriterStub.aborted_messages, [])
get_transaction().commit()
self.assertEquals(MaildirWriterStub.commited_messages,
- [opt_headers + message])
+ [zope_headers + opt_headers + message])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
MaildirWriterStub.commited_messages = []
@@ -183,6 +190,8 @@
self.assert_(MaildirWriterStub.commited_messages[0].endswith(message))
new_headers = MaildirWriterStub.commited_messages[0][:-len(message)]
self.assert_(new_headers.find('Message-Id: <%s>' % msgid) != -1)
+ self.assert_(new_headers.find('X-Zope-From: %s' % fromaddr) != 1)
+ self.assert_(new_headers.find('X-Zope-To: %s' % ", ".join(toaddrs)) != 1)
self.assertEquals(MaildirWriterStub.aborted_messages, [])
MaildirWriterStub.commited_messages = []
@@ -194,9 +203,44 @@
self.assertEquals(len(MaildirWriterStub.aborted_messages), 1)
+class TestQueueProcessorThread(TestCase):
+
+ def setUp(self):
+ from zope.app.mail.service import QueueProcessorThread
+ self.md = MaildirStub('/foo/bar/baz')
+ self.thread = QueueProcessorThread()
+ self.thread.setMaildir(self.md)
+ self.mailer = MailerStub()
+ self.thread.setMailer(self.mailer)
+
+ def test_parseMessage(self):
+
+ hdr = ('X-Zope-From: foo@example.com\n'
+ 'X-Zope-To: bar@example.com, baz@example.com\n')
+ msg = ('Header: value\n'
+ '\n'
+ 'Body\n')
+
+
+ f, t, m = self.thread._parseMessage(hdr + msg)
+ self.assertEquals(f, 'foo@example.com')
+ self.assertEquals(t, ('bar@example.com', 'baz@example.com'))
+ self.assertEquals(m, msg)
+
+ def test_deliveration(self):
+ self.md.msgs.append('X-Zope-From: foo@example.com\n'
+ 'X-Zope-To: bar@example.com, baz@example.com\n'
+ 'Header: value\n\nBody\n')
+ self.thread.run(forever=False)
+ self.assertEquals(self.mailer.sent_messages,
+ [('foo@example.com',
+ ('bar@example.com', 'baz@example.com'),
+ 'Header: value\n\nBody\n')])
+
def test_suite():
return TestSuite((
makeSuite(TestMailDataManager),
makeSuite(TestDirectMailService),
makeSuite(TestQueuedMailService),
+ makeSuite(TestQueueProcessorThread),
))