Merge bounce cleanup branch. This is a major cleansing and refactoring of the

bounce processor.  Note that one thing this does not do is implement any of
the policy around bounce scores.  It "merely" cleans up all the crud in the
BounceRunner, and all the logic around registering bounce events.  This means
the bounce runner can actually be enabled again.  More code needs to be
written to process bounces events occasionally.

Details:

* maybe_forward() moved to app/bounces.py, and it can now discard the
  message, forward it to the list administrators (moderators + owners), or
  site owners.  See UnrecognizedBounceDisposition.
* scan_message() always returns a set of addresses.
* The DSN bounce detector is cleaned up.
* An IBounceProcessor utility is added.
* IBounceEvents are added, with database support.
* bounce_unrecognized_goes_to_list_owner -> forward_unrecognized_bounces_to
* bounce_processing -> process_bounces
* ReopenableFileHandler.filename is exposed as a public attribute.  This aids
  in testing.
* Fix the signature of UserNotification.__init__() to be more PEP 8 compliant.
* Change the OwnerNotification.__init__() signature to take a roster object
  instead of `tomoderators`.  When the roster is None, the site owner is used
  instead.
* Fix the default style setting; s/personalization/personalize/
* BounceMixin is gone, baby gone.
* Add tests for the OutgoingRunner.
* make_testable_runner() can now take a predicate object, which can change how
  often the runner goes through its main loop.  Use this e.g. to go through
  only once when a message does not get dequeued.
* A new LogFileMark helper for testing messages to a log file.

3 files changed, 785 insertions, 0 deletions

diff --git a/src/mailman/queue/tests/__init__.py b/src/mailman/queue/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/src/mailman/queue/tests/__init__.py
diff --git a/src/mailman/queue/tests/test_bounce.py b/src/mailman/queue/tests/test_bounce.py
new file mode 100644
index 000000000..1946df50c
--- /dev/null
+++ b/src/mailman/queue/tests/test_bounce.py
@@ -0,0 +1,236 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the bounce queue runner."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'test_suite',
+    ]
+
+
+import unittest
+
+from zope.component import getUtility
+
+from mailman.app.bounces import send_probe
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.bounce import (
+    BounceContext, IBounceProcessor, UnrecognizedBounceDisposition)
+from mailman.interfaces.member import MemberRole
+from mailman.interfaces.usermanager import IUserManager
+from mailman.queue.bounce import BounceRunner
+from mailman.testing.helpers import (
+    LogFileMark,
+    get_queue_messages,
+    make_testable_runner,
+    specialized_message_from_string as message_from_string)
+from mailman.testing.layers import ConfigLayer
+
+
+
+class TestBounceQueue(unittest.TestCase):
+    """Test the bounce queue runner."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._bounceq = config.switchboards['bounces']
+        self._runner = make_testable_runner(BounceRunner, 'bounces')
+        self._anne = getUtility(IUserManager).create_address(
+            'anne@example.com')
+        self._member = self._mlist.subscribe(self._anne, MemberRole.member)
+        self._msg = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces+anne=example.com@example.com
+Message-Id: <first>
+
+""")
+        self._msgdata = dict(listname='test@example.com')
+        self._processor = getUtility(IBounceProcessor)
+        config.push('site owner', """
+        [mailman]
+        site_owner: postmaster@example.com
+        """)
+
+    def tearDown(self):
+        config.pop('site owner')
+
+    def test_does_no_processing(self):
+        # If the mailing list does no bounce processing, the messages are
+        # simply discarded.
+        self._mlist.bounce_processing = False
+        self._bounceq.enqueue(self._msg, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        self.assertEqual(len(list(self._processor.events)), 0)
+
+    def test_verp_detection(self):
+        # When we get a VERPd bounce, and we're doing processing, a bounce
+        # event will be registered.
+        self._bounceq.enqueue(self._msg, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<first>')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonfatal_verp_detection(self):
+        # A VERPd bounce was received, but the error was nonfatal.
+        nonfatal = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces+anne=example.com@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: delayed
+Original-Recipient: rfc822; somebody@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(nonfatal, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+
+    def test_verp_probe_bounce(self):
+        # A VERP probe bounced.  The primary difference here is that the
+        # registered bounce event will have a different context.  The
+        # Message-Id will be different too, because of the way we're
+        # simulating the probe bounce.
+        #
+        # Start be simulating a probe bounce.
+        send_probe(self._member, self._msg)
+        message = get_queue_messages('virgin')[0].msg
+        bounce = message_from_string("""\
+To: {0}
+From: mail-daemon@example.com
+Message-Id: <second>
+
+""".format(message['From']))
+        self._bounceq.enqueue(bounce, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<second>')
+        self.assertEqual(events[0].context, BounceContext.probe)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonverp_detectable_fatal_bounce(self):
+        # Here's a bounce that is not VERPd, but which has a bouncing address
+        # that can be parsed from a known bounce format.  DSN is as good as
+        # any, but we'll make the parsed address different for the fun of it.
+        dsn = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: fail
+Original-Recipient: rfc822; bart@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(dsn, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'bart@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<first>')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonverp_detectable_nonfatal_bounce(self):
+        # Here's a bounce that is not VERPd, but which has a bouncing address
+        # that can be parsed from a known bounce format.  The bounce is
+        # non-fatal so no bounce event is registered.
+        dsn = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: delayed
+Original-Recipient: rfc822; bart@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(dsn, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+
+    def test_no_detectable_bounce_addresses(self):
+        # A bounce message was received, but no addresses could be detected.
+        # A message will be logged in the bounce log though, and the message
+        # can be forwarded to someone who can do something about it.
+        self._mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.site_owner)
+        bogus = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <third>
+
+""")
+        self._bounceq.enqueue(bogus, self._msgdata)
+        mark = LogFileMark('mailman.bounce')
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+        line = mark.readline()
+        self.assertEqual(
+            line[-51:-1],
+            'Bounce message w/no discernable addresses: <third>')
+        # Here's the forwarded message to the site owners.
+        forwards = get_queue_messages('virgin')
+        self.assertEqual(len(forwards), 1)
+        self.assertEqual(forwards[0].msg['to'], 'postmaster@example.com')
+
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestBounceQueue))
+    return suite
diff --git a/src/mailman/queue/tests/test_outgoing.py b/src/mailman/queue/tests/test_outgoing.py
new file mode 100644
index 000000000..a0fe407c8
--- /dev/null
+++ b/src/mailman/queue/tests/test_outgoing.py
@@ -0,0 +1,549 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the outgoing queue runner."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'test_suite',
+    ]
+
+
+import os
+import socket
+import logging
+import unittest
+
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+from lazr.config import as_timedelta
+from zope.component import getUtility
+
+from mailman.app.bounces import send_probe
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.bounce import BounceContext, IBounceProcessor
+from mailman.interfaces.mailinglist import Personalization
+from mailman.interfaces.member import MemberRole
+from mailman.interfaces.mta import SomeRecipientsFailed
+from mailman.interfaces.pending import IPendings
+from mailman.interfaces.usermanager import IUserManager
+from mailman.queue.outgoing import OutgoingRunner
+from mailman.testing.helpers import (
+    LogFileMark,
+    get_queue_messages,
+    make_testable_runner,
+    specialized_message_from_string as message_from_string)
+from mailman.testing.layers import ConfigLayer, SMTPLayer
+from mailman.utilities.datetime import factory, now
+
+
+
+def run_once(qrunner):
+    """Predicate for make_testable_runner().
+
+    Ensures that the queue runner only runs once.
+    """
+    return True
+
+
+@contextmanager
+def temporary_config(name, settings):
+    """Temporarily set a configuration (use in a with-statement)."""
+    config.push(name, settings)
+    try:
+        yield
+    finally:
+        config.pop(name)
+
+
+
+class TestOnce(unittest.TestCase):
+    """Test outgoing runner message disposition."""
+
+    layer = SMTPLayer
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+        self._msgdata = {}
+
+    def test_deliver_after(self):
+        # When the metadata has a deliver_after key in the future, the queue
+        # runner will re-enqueue the message rather than delivering it.
+        deliver_after = now() + timedelta(days=10)
+        self._msgdata['deliver_after'] = deliver_after
+        self._outq.enqueue(self._msg, self._msgdata,
+                           tolist=True, listname='test@example.com')
+        self._runner.run()
+        items = get_queue_messages('out')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['deliver_after'], deliver_after)
+        self.assertEqual(items[0].msg['message-id'], '<first>')
+
+
+
+captured_mlist = None
+captured_msg = None
+captured_msgdata = None
+
+def capture(mlist, msg, msgdata):
+    global captured_mlist, captured_msg, captured_msgdata
+    captured_mlist = mlist
+    captured_msg = msg
+    captured_msgdata = msgdata
+
+
+class TestVERPSettings(unittest.TestCase):
+    """Test the selection of VERP based on various criteria."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        global captured_mlist, captured_msg, captured_msgdata
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.capture
+        """)
+        # Reset the captured data.
+        captured_mlist = None
+        captured_msg = None
+        captured_msgdata = None
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out')
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_delivery_callback(self):
+        # Test that the configuration variable calls the appropriate callback.
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        self.assertEqual(captured_mlist, self._mlist)
+        self.assertEqual(captured_msg.as_string(), self._msg.as_string())
+        # Of course, the message metadata will contain a bunch of keys added
+        # by the processing.  We don't really care about the details, so this
+        # test is a good enough stand-in.
+        self.assertEqual(captured_msgdata['listname'], 'test@example.com')
+
+    def test_verp_in_metadata(self):
+        # Test that if the metadata has a 'verp' key, it is unchanged.
+        marker = 'yepper'
+        msgdata = dict(verp=marker)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], marker)
+
+    def test_personalized_individual_deliveries_verp(self):
+        # When deliveries are personalized, and the configuration setting
+        # indicates, messages will be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.individual
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_personalized_deliveries: yes
+        """):
+            self._runner.run()
+        self.assertTrue(captured_msgdata['verp'])
+
+    def test_personalized_full_deliveries_verp(self):
+        # When deliveries are personalized, and the configuration setting
+        # indicates, messages will be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.full
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_personalized_deliveries: yes
+        """):
+            self._runner.run()
+        self.assertTrue(captured_msgdata['verp'])
+
+    def test_personalized_deliveries_no_verp(self):
+        # When deliveries are personalized, but the configuration setting
+        # does not indicate, messages will not be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.full
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        self.assertFalse('verp' in captured_msgdata)
+
+    def test_verp_never(self):
+        # Never VERP when the interval is zero.
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 0
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], False)
+
+    def test_verp_always(self):
+        # Always VERP when the interval is one.
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 1
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], True)
+
+    def test_verp_on_interval_match(self):
+        # VERP every so often, when the post_id matches.
+        self._mlist.post_id = 5
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 5
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], True)
+
+    def test_no_verp_on_interval_miss(self):
+        # VERP every so often, when the post_id matches.
+        self._mlist.post_id = 4
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+            [mta]
+            verp_delivery_interval: 5
+            """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], False)
+
+
+
+def raise_socket_error(mlist, msg, msgdata):
+    raise socket.error
+
+
+class TestSocketError(unittest.TestCase):
+    """Test socket.error occurring in the delivery function."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.raise_socket_error
+        """)
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_error_with_port_0(self):
+        # Test the code path where a socket.error is raised in the delivery
+        # function, and the MTA port is set to zero.  The only real effect of
+        # that is a log message.  Start by opening the error log and reading
+        # the current file position.
+        error_log = logging.getLogger('mailman.error')
+        filename = error_log.handlers[0].filename
+        filepos = os.stat(filename).st_size
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        with temporary_config('port 0', """
+            [mta]
+            smtp_port: 0
+            """):
+            self._runner.run()
+        with open(filename) as fp:
+            fp.seek(filepos)
+            line = fp.readline()
+        # The log line will contain a variable timestamp, the PID, and a
+        # trailing newline.  Ignore these.
+        self.assertEqual(
+            line[-53:-1],
+            'Cannot connect to SMTP server localhost on port smtp')
+
+    def test_error_with_numeric_port(self):
+        # Test the code path where a socket.error is raised in the delivery
+        # function, and the MTA port is set to zero.  The only real effect of
+        # that is a log message.  Start by opening the error log and reading
+        # the current file position.
+        mark = LogFileMark('mailman.error')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        with temporary_config('port 0', """
+            [mta]
+            smtp_port: 2112
+            """):
+            self._runner.run()
+        line = mark.readline()
+        # The log line will contain a variable timestamp, the PID, and a
+        # trailing newline.  Ignore these.
+        self.assertEqual(
+            line[-53:-1],
+            'Cannot connect to SMTP server localhost on port 2112')
+
+
+
+temporary_failures = []
+permanent_failures = []
+
+
+def raise_SomeRecipientsFailed(mlist, msg, msgdata):
+    raise SomeRecipientsFailed(temporary_failures, permanent_failures)
+
+
+class TestSomeRecipientsFailed(unittest.TestCase):
+    """Test socket.error occurring in the delivery function."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        global temporary_failures, permanent_failures
+        del temporary_failures[:]
+        del permanent_failures[:]
+        self._processor = getUtility(IBounceProcessor)
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.raise_SomeRecipientsFailed
+        """)
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_probe_failure(self):
+        # When a probe message fails during SMTP, a bounce event is recorded
+        # with the proper bounce context.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        msgdata = dict(probe_token=token)
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 1)
+        event = events[0]
+        self.assertEqual(event.list_name, 'test@example.com')
+        self.assertEqual(event.email, 'anne@example.com')
+        self.assertEqual(event.timestamp, datetime(2005, 8, 1, 7, 49, 23))
+        self.assertEqual(event.message_id, '<first>')
+        self.assertEqual(event.context, BounceContext.probe)
+        self.assertEqual(event.processed, False)
+
+    def test_confirmed_probe_failure(self):
+        # This time, a probe also fails, but for some reason the probe token
+        # has already been confirmed and no longer exists in the database.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        getUtility(IPendings).confirm(token)
+        msgdata = dict(probe_token=token)
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+
+    def test_probe_temporary_failure(self):
+        # This time, a probe also fails, but the failures are temporary so
+        # they are not registered.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        getUtility(IPendings).confirm(token)
+        msgdata = dict(probe_token=token)
+        temporary_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+
+    def test_one_permanent_failure(self):
+        # Normal (i.e. non-probe) permanent failures just get registered.
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+
+    def test_two_permanent_failures(self):
+        # Two normal (i.e. non-probe) permanent failures just get registered.
+        permanent_failures.append('anne@example.com')
+        permanent_failures.append('bart@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[1].email, 'bart@example.com')
+        self.assertEqual(events[1].context, BounceContext.normal)
+
+    def test_one_temporary_failure(self):
+        # The first time there are temporary failures, the message just gets
+        # put in the retry queue, but with some metadata to prevent infinite
+        # retries.
+        temporary_failures.append('cris@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(self._msg.as_string(), items[0].msg.as_string())
+        # The metadata has three keys which are used two decide whether the
+        # next temporary failure should be retried.
+        self.assertEqual(items[0].msgdata['last_recip_count'], 1)
+        deliver_until = (datetime(2005, 8, 1, 7, 49, 23) +
+                         as_timedelta(config.mta.delivery_retry_period))
+        self.assertEqual(items[0].msgdata['deliver_until'], deliver_until)
+        self.assertEqual(items[0].msgdata['recipients'], ['cris@example.com'])
+
+    def test_two_temporary_failures(self):
+        # The first time there are temporary failures, the message just gets
+        # put in the retry queue, but with some metadata to prevent infinite
+        # retries.
+        temporary_failures.append('cris@example.com')
+        temporary_failures.append('dave@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+        items = get_queue_messages('retry')
+        # There's still only one item in the retry queue, but the metadata
+        # contains both temporary failures.
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['last_recip_count'], 2)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['cris@example.com', 'dave@example.com'])
+
+    def test_mixed_failures(self):
+        # Some temporary and some permanent failures.
+        permanent_failures.append('elle@example.com')
+        permanent_failures.append('fred@example.com')
+        temporary_failures.append('gwen@example.com')
+        temporary_failures.append('herb@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        # Let's look at the permanent failures.
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0].email, 'elle@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[1].email, 'fred@example.com')
+        self.assertEqual(events[1].context, BounceContext.normal)
+        # Let's look at the temporary failures.
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['gwen@example.com', 'herb@example.com'])
+
+    def test_no_progress_on_retries_within_retry_period(self):
+        # Temporary failures cause queuing for a retry later on, unless no
+        # progress is being made on the retries and we've tried for the
+        # specified delivery retry period.  This test ensures that even if no
+        # progress is made, if the retry period hasn't expired, the message
+        # will be requeued.
+        temporary_failures.append('iona@example.com')
+        temporary_failures.append('jeff@example.com')
+        deliver_until = (datetime(2005, 8, 1, 7, 49, 23) +
+                         as_timedelta(config.mta.delivery_retry_period))
+        msgdata = dict(last_recip_count=2,
+                       deliver_until=deliver_until)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        # The retry queue should have our message waiting to be retried.
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['deliver_until'], deliver_until)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['iona@example.com', 'jeff@example.com'])
+
+    def test_no_progress_on_retries_with_expired_retry_period(self):
+        # We've had temporary failures with no progress, and the retry period
+        # has expired.  In that case, a log entry is written and message is
+        # discarded.  There's nothing more that can be done.
+        temporary_failures.append('kira@example.com')
+        temporary_failures.append('lonn@example.com')
+        retry_period = as_timedelta(config.mta.delivery_retry_period)
+        deliver_until = datetime(2005, 8, 1, 7, 49, 23) + retry_period
+        msgdata = dict(last_recip_count=2,
+                       deliver_until=deliver_until)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        # Before the queue runner runs, several days pass.
+        factory.fast_forward(retry_period.days + 1)
+        mark = LogFileMark('mailman.smtp')
+        self._runner.run()
+        # There should be no message in the retry or outgoing queues.
+        self.assertEqual(len(get_queue_messages('retry')), 0)
+        self.assertEqual(len(get_queue_messages('out')), 0)
+        # There should be a log message in the smtp log indicating that the
+        # message has been discarded.
+        line = mark.readline()
+        self.assertEqual(
+            line[-63:-1],
+            'Discarding message with persistent temporary failures: <first>')
+
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestOnce))
+    suite.addTest(unittest.makeSuite(TestVERPSettings))
+    suite.addTest(unittest.makeSuite(TestSocketError))
+    suite.addTest(unittest.makeSuite(TestSomeRecipientsFailed))
+    return suite