[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