[Zope3-checkins] CVS: Zope3/src/zope/app/mail/tests - test_event.py:1.1.2.1 test_service.py:1.1.2.1 test_directives.py:1.3.2.1 test_asyncmailservice.py:NONE test_batchmailer.py:NONE test_mailevents.py:NONE test_simplemailer.py:NONE
Viktorija Zaksiene
ryzaja@codeworks.lt
Thu, 22 May 2003 04:51:43 -0400
Update of /cvs-repository/Zope3/src/zope/app/mail/tests
In directory cvs.zope.org:/tmp/cvs-serv25415/src/zope/app/mail/tests
Modified Files:
Tag: cw-mail-branch
test_directives.py
Added Files:
Tag: cw-mail-branch
test_event.py test_service.py
Removed Files:
Tag: cw-mail-branch
test_asyncmailservice.py test_batchmailer.py
test_mailevents.py test_simplemailer.py
Log Message:
Marius and Viktorija.
Refactored Mail delivery service. The work is still not finished, so we
commit it to the branch.
=== Added File Zope3/src/zope/app/mail/tests/test_event.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 MailService, Mailers and MailEvents.
$Id: test_event.py,v 1.1.2.1 2003/05/22 08:51:41 ryzaja Exp $
"""
from unittest import TestCase, TestSuite, makeSuite
from zope.interface.verify import verifyObject
from zope.app.interfaces.mail import IMailSentEvent, IMailErrorEvent
class TestMailEvents(TestCase):
def testMailSendEvent(self):
from zope.app.mail.event import MailSentEvent
msgid = '<1234@example.com>'
m = MailSentEvent(msgid)
verifyObject(IMailSentEvent, m)
self.assertEquals(m.messageId, msgid)
def testMailErrorEvent(self):
from zope.app.mail.event import MailErrorEvent
msgid = '<1234@example.com>'
error = '550 Relay access denied'
m = MailErrorEvent(msgid, error)
verifyObject(IMailErrorEvent, m)
self.assertEquals(m.messageId, msgid)
self.assertEquals(m.errorMessage, error)
def test_suite():
return TestSuite((
makeSuite(TestMailEvents),
))
=== Added File Zope3/src/zope/app/mail/tests/test_service.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 MailService, Mailers and MailEvents.
$Id: test_service.py,v 1.1.2.1 2003/05/22 08:51:41 ryzaja Exp $
"""
from unittest import TestCase, TestSuite, makeSuite
from zope.interface import implements
from zope.interface.verify import verifyObject
from zope.app.interfaces.mail import IMailer
from transaction import get_transaction
__metaclass__ = type
class MailerStub:
implements(IMailer)
sent_messages = ()
def send(self, fromaddr, toaddrs, message):
self.sent_messages += ((fromaddr, toaddrs, message), )
class TestMailDataManager(TestCase):
def testInterface(self):
from transaction.interfaces import IDataManager
from zope.app.mail.service import MailDataManager
manager = MailDataManager(object, (1, 2))
verifyObject(IDataManager, manager)
self.assertEqual(manager.callable, object)
self.assertEqual(manager.args, (1, 2))
class TestDirectMailService(TestCase):
def testInterface(self):
from zope.app.interfaces.mail import IDirectMailService
from zope.app.mail.service import DirectMailService
mailer = MailerStub()
service = DirectMailService(mailer)
verifyObject(IDirectMailService, service)
self.assertEqual(service.mailer, mailer)
def testSend(self):
from zope.app.mail.service import DirectMailService
mailer = MailerStub()
service = DirectMailService(mailer)
fromaddr = 'Jim <jim@example.com'
toaddrs = ('Guido <guido@example.com>',
'Steve <steve@examplecom>')
opt_headers = ('From: Jim <jim@example.org>\n'
'To: some-zope-coders:;\n'
'Date: Mon, 19 May 2003 10:17:36 -0400\n'
'Message-Id: <20030519.1234@example.org>\n')
message = ('Subject: example\n'
'\n'
'This is just an example\n')
msgid = service.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(msgid, '20030519.1234@example.org')
self.assertEquals(mailer.sent_messages, ())
get_transaction().commit()
self.assertEquals(mailer.sent_messages,
((fromaddr, toaddrs, opt_headers + message), ))
mailer.sent_messages = ()
msgid = service.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 = service.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(mailer.sent_messages, ())
get_transaction().abort()
self.assertEquals(mailer.sent_messages, ())
class MaildirWriterStub:
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:
def __init__(self, path, create=False):
self.path = path
self.create = create
def __iter__(self):
return iter(())
def newMessage(self):
return MaildirWriterStub()
class TestQueuedMailService(TestCase):
def setUp(self):
import zope.app.mail.service as mail_service_module
self.mail_service_module = mail_service_module
self.old_Maildir = mail_service_module.Maildir
mail_service_module.Maildir = MaildirStub
def tearDown(self):
self.mail_service_module.Maildir = self.old_Maildir
def testInterface(self):
from zope.app.interfaces.mail import IQueuedMailService
from zope.app.mail.service import QueuedMailService
service = QueuedMailService('/path/to/mailbox')
verifyObject(IQueuedMailService, service)
self.assertEqual(service.queuePath, '/path/to/mailbox')
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>')
opt_headers = ('From: Jim <jim@example.org>\n'
'To: some-zope-coders:;\n'
'Date: Mon, 19 May 2003 10:17:36 -0400\n'
'Message-Id: <20030519.1234@example.org>\n')
message = ('Subject: example\n'
'\n'
'This is just an example\n')
msgid = service.send(fromaddr, toaddrs, opt_headers + message)
self.assertEquals(msgid, '20030519.1234@example.org')
self.assertEquals(MaildirWriterStub.commited_messages, [])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
get_transaction().commit()
self.assertEquals(MaildirWriterStub.commited_messages,
[opt_headers + message])
self.assertEquals(MaildirWriterStub.aborted_messages, [])
MaildirWriterStub.commited_messages = []
msgid = service.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.assertEquals(MaildirWriterStub.aborted_messages, [])
MaildirWriterStub.commited_messages = []
msgid = service.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)
def test_suite():
return TestSuite((
makeSuite(TestMailDataManager),
makeSuite(TestDirectMailService),
makeSuite(TestQueuedMailService),
))
=== Zope3/src/zope/app/mail/tests/test_directives.py 1.3 => 1.3.2.1 ===
--- Zope3/src/zope/app/mail/tests/test_directives.py:1.3 Mon May 19 06:03:37 2003
+++ Zope3/src/zope/app/mail/tests/test_directives.py Thu May 22 04:51:41 2003
@@ -26,6 +26,7 @@
from zope.app.component.metaconfigure import managerHandler, provideInterface
import zope.app.mail
import zope.app.interfaces.mail
+from zope.app.mail.metaconfigure import provideMailer
template = """<zopeConfigure
xmlns='http://namespaces.zope.org/zope'
@@ -46,38 +47,31 @@
XMLConfig('metameta.zcml', zope.configuration)()
XMLConfig('meta.zcml', zope.app.mail)()
- def test_mailservice(self):
+ def testQueuedService(self):
xmlconfig(StringIO(template % (
'''
- <mail:mailservice name="Mail"
- hostname="somehost" port="125"
- username="foo" password="bar"
- class=".mail.AsyncMailService"
+ <mail:queuedService name="Mail"
+ queuePath="/path/to/mailbox"
permission="zope.Public" />
'''
)), None, Context([], zope.app.mail))
service = getService(None, 'Mail')
- self.assertEqual('AsyncMailService', service.__class__.__name__)
- self.assertEqual('somehost', service.hostname)
- self.assertEqual(125, service.port)
- self.assertEqual('foo', service.username)
- self.assertEqual('bar', service.password)
+ self.assertEqual('QueuedMailService', service.__class__.__name__)
+ self.assertEqual('/path/to/mailbox', service.queuePath)
- def test_mailer(self):
+ def testDirectService(self):
+ testMailer = object()
+ provideMailer('test.mailer', testMailer)
xmlconfig(StringIO(template % (
'''
- <mail:mailservice class=".mail.AsyncMailService"
- permission="zope.Public" />
-
- <mail:mailer name="TestSimpleMailer" class=".mailer.SimpleMailer"
- serviceType="Mail" default="True" />
+ <mail:directService name="Mail"
+ mailer="test.mailer"
+ permission="zope.Public" />
'''
)), None, Context([], zope.app.mail))
-
- service = getService(None, "Mail")
- self.assertEqual("TestSimpleMailer", service.getDefaultMailerName())
- self.assertEqual(service._AsyncMailService__mailers['TestSimpleMailer'],
- service.createMailer('TestSimpleMailer').__class__)
+ service = getService(None, 'Mail')
+ self.assertEqual('DirectMailService', service.__class__.__name__)
+ self.assert_(testMailer is service.mailer)
def test_suite():
=== Removed File Zope3/src/zope/app/mail/tests/test_asyncmailservice.py ===
=== Removed File Zope3/src/zope/app/mail/tests/test_batchmailer.py ===
=== Removed File Zope3/src/zope/app/mail/tests/test_mailevents.py ===
=== Removed File Zope3/src/zope/app/mail/tests/test_simplemailer.py ===