src/mailman/chains/tests/test_owner.py

   1 # Copyright (C) 2012-2019 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 import unittest
  21
  22 from mailman.app.lifecycle import create_list
  23 from mailman.chains.owner import BuiltInOwnerChain
  24 from mailman.core.chains import process
  25 from mailman.interfaces.chain import AcceptOwnerEvent
  26 from mailman.testing.helpers import (
  27     event_subscribers, get_queue_messages,
  28     specialized_message_from_string as mfs)
  29 from mailman.testing.layers import ConfigLayer
  30
  31
  32 class TestOwnerChain(unittest.TestCase):
  33     """Test the owner chain."""
  34
  35     layer = ConfigLayer
  36
  37     def setUp(self):
  38         self._mlist = create_list('test@example.com')
  39         self._msg = mfs("""\
  40 From: anne@example.com
  41 To: test@example.com
  42 Message-ID: <ant>
  43
  44 """)
  45
  46     def test_owner_pipeline(self):
  47         # Messages processed through the default owners chain end up in the
  48         # pipeline queue, and an event gets sent.
  49         #
  50         # This event subscriber records the event that occurs when the message
  51         # is processed by the owner chain.
  52         events = []
  53         def catch_event(event):                                  # noqa: E306
  54             if isinstance(event, AcceptOwnerEvent):
  55                 events.append(event)
  56         with event_subscribers(catch_event):
  57             process(self._mlist, self._msg, {}, 'default-owner-chain')
  58         self.assertEqual(len(events), 1)
  59         event = events[0]
  60         self.assertIsInstance(event, AcceptOwnerEvent)
  61         self.assertEqual(event.mlist, self._mlist)
  62         self.assertEqual(event.msg['message-id'], '<ant>')
  63         self.assertIsInstance(event.chain, BuiltInOwnerChain)
  64         items = get_queue_messages('pipeline', expected_count=1)
  65         message = items[0].msg
  66         self.assertEqual(message['message-id'], '<ant>')