[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),
         ))