src/mailman/chains/tests/test_owner.py

   1 # Copyright (C) 2012-2014 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 """Test the owner chain."""
  19
  20 from __future__ import absolute_import, print_function, unicode_literals
  21
  22 __metaclass__ = type
  23 __all__ = [
  24     'TestOwnerChain',
  25     ]
  26
  27
  28 import unittest
  29
  30 from mailman.app.lifecycle import create_list
  31 from mailman.chains.owner import BuiltInOwnerChain
  32 from mailman.core.chains import process
  33 from mailman.interfaces.chain import AcceptOwnerEvent
  34 from mailman.testing.helpers import (
  35     event_subscribers,
  36     get_queue_messages,
  37     specialized_message_from_string as mfs)
  38 from mailman.testing.layers import ConfigLayer
  39
  40
  41 \f
  42 class TestOwnerChain(unittest.TestCase):
  43     """Test the owner chain."""
  44
  45     layer = ConfigLayer
  46
  47     def setUp(self):
  48         self._mlist = create_list('test@example.com')
  49         self._msg = mfs("""\
  50 From: anne@example.com
  51 To: test@example.com
  52 Message-ID: <ant>
  53
  54 """)
  55
  56     def test_owner_pipeline(self):
  57         # Messages processed through the default owners chain end up in the
  58         # pipeline queue, and an event gets sent.
  59         #
  60         # This event subscriber records the event that occurs when the message
  61         # is processed by the owner chain.
  62         events = []
  63         def catch_event(event):
  64             if isinstance(event, AcceptOwnerEvent):
  65                 events.append(event)
  66         with event_subscribers(catch_event):
  67             process(self._mlist, self._msg, {}, 'default-owner-chain')
  68         self.assertEqual(len(events), 1)
  69         event = events[0]
  70         self.assertTrue(isinstance(event, AcceptOwnerEvent))
  71         self.assertEqual(event.mlist, self._mlist)
  72         self.assertEqual(event.msg['message-id'], '<ant>')
  73         self.assertTrue(isinstance(event.chain, BuiltInOwnerChain))
  74         messages = get_queue_messages('pipeline')
  75         self.assertEqual(len(messages), 1)
  76         message = messages[0].msg
  77         self.assertEqual(message['message-id'], '<ant>')