[Zope3-checkins] CVS: Zope3/src/zope/app/mail/tests -
test_delivery.py:1.1 mail.zcml:1.2 test_directives.py:1.10
test_event.py:1.3 test_maildir.py:1.7 test_mailer.py:1.3
test_service.py:NONE
Stephan Richter
srichter at cosmos.phy.tufts.edu
Wed Mar 3 04:15:44 EST 2004
Update of /cvs-repository/Zope3/src/zope/app/mail/tests
In directory cvs.zope.org:/tmp/cvs-serv25620/src/zope/app/mail/tests
Modified Files:
mail.zcml test_directives.py test_event.py test_maildir.py
test_mailer.py
Added Files:
test_delivery.py
Removed Files:
test_service.py
Log Message:
- Renamed MailService to MailDelivery
- Made MailDelivery a utility
- Made Mailer a utility. Removed custom registry.
=== Added File Zope3/src/zope/app/mail/tests/test_delivery.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""MailService Implementation
Simple implementation of the MailDelivery, Mailers and MailEvents.
$Id: test_delivery.py,v 1.1 2004/03/03 09:15:43 srichter Exp $
"""
import os.path
from tempfile import mktemp
from unittest import TestCase, TestSuite, makeSuite
from transaction import get_transaction
from zope.interface import implements
from zope.interface.verify import verifyObject
from zope.app.mail.interfaces import IMailer
class MailerStub(object):
implements(IMailer)
def __init__(self, *args, **kw):
self.sent_messages = []
def send(self, fromaddr, toaddrs, message):
self.sent_messages.append((fromaddr, toaddrs, message))
class TestMailDataManager(TestCase):
def testInterface(self):
from transaction.interfaces import IDataManager
from zope.app.mail.delivery import MailDataManager
manager = MailDataManager(object, (1, 2))
verifyObject(IDataManager, manager)
self.assertEqual(manager.callable, object)
self.assertEqual(manager.args, (1, 2))
class TestDirectMailDelivery(TestCase):
def testInterface(self):
from zope.app.mail.interfaces import IDirectMailDelivery
from zope.app.mail.delivery import DirectMailDelivery
mailer = MailerStub()
delivery = DirectMailDelivery(mailer)
verifyObject(IDirectMailDelivery, delivery)
self.assertEqual(delivery.mailer, mailer)
def testSend(self):
from zope.app.mail.delivery import DirectMailDelivery
mailer = MailerStub()
delivery = DirectMailDelivery(mailer)
fromaddr = 'Jim <jim at example.com'
toaddrs = ('Guido <guido at example.com>',
'Steve <steve at examplecom>')
opt_headers = ('From: Jim <jim at example.org>\n'
'To: some-zope-coders:;\n'
'Date: Mon, 19 May 2003 10:17:36 -0400\n'
'Message-Id: <20030519.1234 at example.org>\n')
message = ('Subject: example\n'
'\n'
'This is just an example\n')
msgid = delivery.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(msgid, '20030519.1234 at example.org')
self.assertEquals(mailer.sent_messages, [])
get_transaction().commit()
self.assertEquals(mailer.sent_messages,
[(fromaddr, toaddrs, opt_headers + message)])
mailer.sent_messages = []
msgid = delivery.send(fromaddr, toaddrs, message)
self.assert_('@' in msgid)
self.assertEquals(mailer.sent_messages, [])
get_transaction().commit()
self.assertEquals(len(mailer.sent_messages), 1)
self.assertEquals(mailer.sent_messages[0][0], fromaddr)
self.assertEquals(mailer.sent_messages[0][1], toaddrs)
self.assert_(mailer.sent_messages[0][2].endswith(message))
new_headers = mailer.sent_messages[0][2][:-len(message)]
self.assert_(new_headers.find('Message-Id: <%s>' % msgid) != -1)
mailer.sent_messages = []
msgid = delivery.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(mailer.sent_messages, [])
get_transaction().abort()
self.assertEquals(mailer.sent_messages, [])
class MaildirWriterStub(object):
data = ''
commited_messages = [] # this list is shared among all instances
aborted_messages = [] # this one too
def write(self, str):
self.data += str
def writelines(self, seq):
self.data += ''.join(seq)
def commit(self):
self._commited = True
self.commited_messages.append(self.data)
def abort(self):
self._aborted = True
self.aborted_messages.append(self.data)
class MaildirStub(object):
def __init__(self, path, create=False):
self.path = path
self.create = create
self.msgs = []
self.files = []
def __iter__(self):
return iter(self.files)
def newMessage(self):
m = MaildirWriterStub()
self.msgs.append(m)
return m
class LoggerStub(object):
def __init__(self):
self.infos = []
self.errors = []
def getLogger(name):
return self
def error(self, msg, *args, **kwargs):
self.errors.append((msg, args, kwargs))
def info(self, msg, *args, **kwargs):
self.infos.append((msg, args, kwargs))
class BizzarreMailError(IOError):
pass
class BrokenMailerStub(object):
implements(IMailer)
def __init__(self, *args, **kw):
pass
def send(self, fromaddr, toaddrs, message):
raise BizzarreMailError("bad things happened while sending mail")
class TestQueuedMailDelivery(TestCase):
def setUp(self):
import zope.app.mail.delivery as mail_delivery_module
self.mail_delivery_module = mail_delivery_module
self.old_Maildir = mail_delivery_module.Maildir
mail_delivery_module.Maildir = MaildirStub
def tearDown(self):
self.mail_delivery_module.Maildir = self.old_Maildir
MaildirWriterStub.commited_messages = []
MaildirWriterStub.aborted_messages = []
def testInterface(self):
from zope.app.mail.interfaces import IQueuedMailDelivery
from zope.app.mail.delivery import QueuedMailDelivery
delivery = QueuedMailDelivery('/path/to/mailbox')
verifyObject(IQueuedMailDelivery, delivery)
self.assertEqual(delivery.queuePath, '/path/to/mailbox')
def testSend(self):
from zope.app.mail.delivery import QueuedMailDelivery
delivery = QueuedMailDelivery('/path/to/mailbox')
fromaddr = 'jim at example.com'
toaddrs = ('guido at example.com',
'steve at examplecom')
zope_headers = ('X-Zope-From: jim at example.com\n'
'X-Zope-To: guido at example.com, steve at examplecom\n')
opt_headers = ('From: Jim <jim at example.org>\n'
'To: some-zope-coders:;\n'
'Date: Mon, 19 May 2003 10:17:36 -0400\n'
'Message-Id: <20030519.1234 at example.org>\n')
message = ('Subject: example\n'
'\n'
'This is just an example\n')
msgid = delivery.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(msgid, '20030519.1234 at example.org')
self.assertEquals(MaildirWriterStub.commited_messages, [])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
get_transaction().commit()
self.assertEquals(MaildirWriterStub.commited_messages,
[zope_headers + opt_headers + message])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
MaildirWriterStub.commited_messages = []
msgid = delivery.send(fromaddr, toaddrs, message)
self.assert_('@' in msgid)
self.assertEquals(MaildirWriterStub.commited_messages, [])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
get_transaction().commit()
self.assertEquals(len(MaildirWriterStub.commited_messages), 1)
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 = []
msgid = delivery.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(MaildirWriterStub.commited_messages, [])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
get_transaction().abort()
self.assertEquals(MaildirWriterStub.commited_messages, [])
self.assertEquals(len(MaildirWriterStub.aborted_messages), 1)
class TestQueueProcessorThread(TestCase):
def setUp(self):
from zope.app.mail.delivery import QueueProcessorThread
self.md = MaildirStub('/foo/bar/baz')
self.thread = QueueProcessorThread()
self.thread.setMaildir(self.md)
self.mailer = MailerStub()
self.thread.setMailer(self.mailer)
self.thread.log = LoggerStub()
def test_parseMessage(self):
hdr = ('X-Zope-From: foo at example.com\n'
'X-Zope-To: bar at example.com, baz at example.com\n')
msg = ('Header: value\n'
'\n'
'Body\n')
f, t, m = self.thread._parseMessage(hdr + msg)
self.assertEquals(f, 'foo at example.com')
self.assertEquals(t, ('bar at example.com', 'baz at example.com'))
self.assertEquals(m, msg)
def test_deliveration(self):
self.filename = mktemp()
temp = open(self.filename, "w+b")
temp.write('X-Zope-From: foo at example.com\n'
'X-Zope-To: bar at example.com, baz at example.com\n'
'Header: value\n\nBody\n')
temp.close()
self.md.files.append(self.filename)
self.thread.run(forever=False)
self.assertEquals(self.mailer.sent_messages,
[('foo at example.com',
('bar at example.com', 'baz at example.com'),
'Header: value\n\nBody\n')])
self.failIf(os.path.exists(self.filename), 'File exists')
self.assertEquals(self.thread.log.infos,
[('Mail from %s to %s sent.',
('foo at example.com',
'bar at example.com, baz at example.com'),
{})])
def test_error_logging(self):
self.thread.setMailer(BrokenMailerStub())
self.filename = mktemp()
temp = open(self.filename, "w+b")
temp.write('X-Zope-From: foo at example.com\n'
'X-Zope-To: bar at example.com, baz at example.com\n'
'Header: value\n\nBody\n')
temp.close()
self.md.files.append(self.filename)
self.thread.run(forever=False)
self.assertEquals(self.thread.log.errors,
[('Error while sending mail from %s to %s.',
('foo at example.com',
'bar at example.com, baz at example.com'),
{'exc_info': 1})])
def test_suite():
return TestSuite((
makeSuite(TestMailDataManager),
makeSuite(TestDirectMailDelivery),
makeSuite(TestQueuedMailDelivery),
makeSuite(TestQueueProcessorThread),
))
if __name__ == '__main__':
unittest.main()
=== Zope3/src/zope/app/mail/tests/mail.zcml 1.1 => 1.2 ===
--- Zope3/src/zope/app/mail/tests/mail.zcml:1.1 Sat Aug 2 08:30:28 2003
+++ Zope3/src/zope/app/mail/tests/mail.zcml Wed Mar 3 04:15:43 2004
@@ -3,30 +3,30 @@
<include package="zope.app.mail" file="meta.zcml"/>
- <mail:queuedService
+ <mail:queuedDelivery
name="Mail"
queuePath="./mailbox"
mailer="test.smtp"
permission="zope.Public" />
- <mail:directService
+ <mail:directDelivery
name="Mail2"
mailer="test.mailer"
permission="zope.Public" />
<mail:sendmailMailer
- id="Sendmail"
+ name="Sendmail"
command="/usr/lib/sendmail -oem -oi -f %(from)s %(to)s" />
<mail:smtpMailer
- id="smtp"
+ name="smtp"
hostname="localhost"
port="25"
username="zope3"
password="xyzzy"/>
<mail:smtpMailer
- id="smtp2"
+ name="smtp2"
hostname="smarthost"/>
</configure>
=== Zope3/src/zope/app/mail/tests/test_directives.py 1.9 => 1.10 ===
--- Zope3/src/zope/app/mail/tests/test_directives.py:1.9 Thu Nov 27 08:59:21 2003
+++ Zope3/src/zope/app/mail/tests/test_directives.py Wed Mar 3 04:15:43 2004
@@ -20,14 +20,17 @@
import threading
import time
-from zope.app.component.metaconfigure import managerHandler, provideInterface
-from zope.app.interfaces.mail import IMailService, ISMTPMailer, ISendmailMailer
-from zope.app.mail.metaconfigure import provideMailer, queryMailer
-from zope.app.mail.service import QueueProcessorThread
-from zope.app.mail import service
-from zope.component import getService
from zope.component.tests.placelesssetup import PlacelessSetup
from zope.configuration import xmlconfig
+from zope.interface import implements
+
+from zope.app import zapi
+from zope.app.component.metaconfigure import managerHandler, provideInterface
+from zope.app.mail.interfaces import \
+ IMailDelivery, IMailer, ISMTPMailer, ISendmailMailer
+from zope.app.mail.delivery import QueueProcessorThread
+from zope.app.mail import delivery
+from zope.app.tests import ztapi
import zope.app.mail.tests
@@ -43,24 +46,25 @@
def newMessage(self):
return None
+class Mailer:
+ implements(IMailer)
+
class DirectivesTest(PlacelessSetup, unittest.TestCase):
def setUp(self):
super(DirectivesTest, self).setUp()
- managerHandler('defineService', 'Mail', IMailService)
- managerHandler('defineService', 'Mail2', IMailService)
- provideInterface('zope.app.interfaces.mail.IMailService', IMailService)
- provideMailer("test.smtp", object())
- self.testMailer = object()
- provideMailer('test.mailer', self.testMailer)
+ self.testMailer = Mailer()
+
+ ztapi.provideUtility(IMailer, Mailer(), name="test.smtp")
+ ztapi.provideUtility(IMailer, self.testMailer, name="test.mailer")
+
self.context = xmlconfig.file("mail.zcml", zope.app.mail.tests)
- self.orig_maildir = service.Maildir
- service.Maildir = MaildirStub
+ self.orig_maildir = delivery.Maildir
+ delivery.Maildir = MaildirStub
-
def tearDown(self):
- service.Maildir = self.orig_maildir
+ delivery.Maildir = self.orig_maildir
# Tear down the mail queue processor thread.
# Give the other thread a chance to start:
@@ -71,26 +75,25 @@
thread.stop()
thread.join()
-
-
- def testQueuedService(self):
- service = getService(None, 'Mail')
- self.assertEqual('QueuedMailService', service.__class__.__name__)
+ def testQueuedDelivery(self):
+ delivery = zapi.getUtility(None, IMailDelivery, "Mail")
+ self.assertEqual('QueuedMailDelivery', delivery.__class__.__name__)
testdir = os.path.dirname(zope.app.mail.tests.__file__)
self.assertEqual(os.path.join(testdir, 'mailbox'),
- service.queuePath)
-
- def testDirectService(self):
- service = getService(None, 'Mail2')
- self.assertEqual('DirectMailService', service.__class__.__name__)
- self.assert_(self.testMailer is service.mailer)
+ delivery.queuePath)
+ def testDirectDelivery(self):
+ delivery = zapi.getUtility(None, IMailDelivery, "Mail2")
+ self.assertEqual('DirectMailDelivery', delivery.__class__.__name__)
+ self.assert_(self.testMailer is delivery.mailer)
def testSendmailMailer(self):
- self.assert_(ISendmailMailer.isImplementedBy(queryMailer("Sendmail")))
+ mailer = zapi.getUtility(None, IMailer, "Sendmail")
+ self.assert_(ISendmailMailer.isImplementedBy(mailer))
def testSMTPMailer(self):
- self.assert_(ISMTPMailer.isImplementedBy(queryMailer("smtp")))
+ mailer = zapi.getUtility(None, IMailer, "smtp")
+ self.assert_(ISMTPMailer.isImplementedBy(mailer))
def test_suite():
=== Zope3/src/zope/app/mail/tests/test_event.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/mail/tests/test_event.py:1.2 Mon Jun 23 11:45:40 2003
+++ Zope3/src/zope/app/mail/tests/test_event.py Wed Mar 3 04:15:43 2004
@@ -18,14 +18,16 @@
$Id$
"""
from unittest import TestCase, TestSuite, makeSuite
+
from zope.interface.verify import verifyObject
-from zope.app.interfaces.mail import IMailSentEvent, IMailErrorEvent
+
+from zope.app.mail.interfaces import IMailSentEvent, IMailErrorEvent
+from zope.app.mail.event import MailSentEvent
class TestMailEvents(TestCase):
def testMailSendEvent(self):
- from zope.app.mail.event import MailSentEvent
msgid = '<1234 at example.com>'
m = MailSentEvent(msgid)
verifyObject(IMailSentEvent, m)
@@ -45,3 +47,6 @@
return TestSuite((
makeSuite(TestMailEvents),
))
+
+if __name__ == '__main__':
+ unittest.main()
=== Zope3/src/zope/app/mail/tests/test_maildir.py 1.6 => 1.7 ===
--- Zope3/src/zope/app/mail/tests/test_maildir.py:1.6 Tue Sep 23 18:16:13 2003
+++ Zope3/src/zope/app/mail/tests/test_maildir.py Wed Mar 3 04:15:43 2004
@@ -15,20 +15,18 @@
$Id$
"""
-
import unittest
import stat
from zope.interface.verify import verifyObject
-__metaclass__ = type
-class FakeSocketModule:
+class FakeSocketModule(object):
def gethostname(self):
return 'myhostname'
-class FakeTimeModule:
+class FakeTimeModule(object):
_timer = 1234500000
@@ -38,7 +36,7 @@
def sleep(self, n):
self._timer += n
-class FakeOsPathModule:
+class FakeOsPathModule(object):
def __init__(self, files, dirs):
self.files = files
@@ -55,7 +53,7 @@
def exists(self, p):
return self._exists_never_fails or p in self.files
-class FakeOsModule:
+class FakeOsModule(object):
F_OK = 0
_stat_mode = {
@@ -109,7 +107,7 @@
def rename(self, old, new):
self._renamed_files += ((old, new), )
-class FakeFile:
+class FakeFile(object):
def __init__(self, filename, mode):
self._filename = filename
@@ -152,7 +150,7 @@
self.fake_os_module.path._exists_never_fails = False
def test_factory(self):
- from zope.app.interfaces.mail import IMaildirFactory, IMaildir
+ from zope.app.mail.interfaces import IMaildirFactory, IMaildir
from zope.app.mail.maildir import Maildir
verifyObject(IMaildirFactory, Maildir)
@@ -193,7 +191,7 @@
def test_newMessage(self):
from zope.app.mail.maildir import Maildir
- from zope.app.interfaces.mail import IMaildirMessageWriter
+ from zope.app.mail.interfaces import IMaildirMessageWriter
m = Maildir('/path/to/maildir')
fd = m.newMessage()
verifyObject(IMaildirMessageWriter, fd)
@@ -202,7 +200,7 @@
def test_newMessage_never_loops(self):
from zope.app.mail.maildir import Maildir
- from zope.app.interfaces.mail import IMaildirMessageWriter
+ from zope.app.mail.interfaces import IMaildirMessageWriter
self.fake_os_module.path._exists_never_fails = True
m = Maildir('/path/to/maildir')
self.assertRaises(RuntimeError, m.newMessage)
=== Zope3/src/zope/app/mail/tests/test_mailer.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/mail/tests/test_mailer.py:1.2 Mon Jun 23 11:45:40 2003
+++ Zope3/src/zope/app/mail/tests/test_mailer.py Wed Mar 3 04:15:43 2004
@@ -19,7 +19,7 @@
import unittest
from StringIO import StringIO
from zope.interface.verify import verifyObject
-from zope.app.interfaces.mail import ISendmailMailer, ISMTPMailer
+from zope.app.mail.interfaces import ISendmailMailer, ISMTPMailer
class TestSendmailMailer(unittest.TestCase):
@@ -43,7 +43,8 @@
def test_send(self):
msgtext = 'Headers: headers\n\nbodybodybody\n-- \nsig\n'
- self.mailer.send('me at example.com', ('you at example.com', 'him at example.com'),
+ self.mailer.send('me at example.com',
+ ('you at example.com', 'him at example.com'),
msgtext)
self.assertEquals(self.input.getvalue(), msgtext)
self.assertEquals(self.cmd_arg, "/usr/lib/sendmail -oem -oi"
=== Removed File Zope3/src/zope/app/mail/tests/test_service.py ===
More information about the Zope3-Checkins
mailing list