src/mailman/runners/tests/test_incoming.py

   1 # Copyright (C) 2012-2016 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 incoming queue runner."""
  19
  20 import unittest
  21
  22 from mailman.app.lifecycle import create_list
  23 from mailman.chains.base import TerminalChainBase
  24 from mailman.config import config
  25 from mailman.runners.incoming import IncomingRunner
  26 from mailman.testing.helpers import (
  27     get_queue_messages, make_testable_runner,
  28     specialized_message_from_string as mfs)
  29 from mailman.testing.layers import ConfigLayer
  30
  31
  32 class Chain(TerminalChainBase):
  33     name = 'test'
  34     description = 'a test chain'
  35
  36     def __init__(self, marker):
  37         self._marker = marker
  38
  39     def _process(self, mlist, msg, msgdata):
  40         msgdata['marker'] = self._marker
  41         config.switchboards['out'].enqueue(msg, msgdata)
  42
  43
  44 class TestIncoming(unittest.TestCase):
  45     """Test the incoming queue runner."""
  46
  47     layer = ConfigLayer
  48
  49     def setUp(self):
  50         self._mlist = create_list('test@example.com')
  51         self._mlist.posting_chain = 'test posting'
  52         self._mlist.owner_chain = 'test owner'
  53         config.chains['test posting'] = Chain('posting')
  54         self.addCleanup(config.chains.pop, 'test posting')
  55         config.chains['test owner'] = Chain('owner')
  56         self.addCleanup(config.chains.pop, 'test owner')
  57         self._in = make_testable_runner(IncomingRunner, 'in')
  58         self._msg = mfs("""\
  59 From: anne@example.com
  60 To: test@example.com
  61
  62 """)
  63
  64     def test_posting(self):
  65         # A message posted to the list goes through the posting chain.
  66         msgdata = dict(listid='test.example.com')
  67         config.switchboards['in'].enqueue(self._msg, msgdata)
  68         self._in.run()
  69         items = get_queue_messages('out', expected_count=1)
  70         self.assertEqual(items[0].msgdata.get('marker'), 'posting')
  71
  72     def test_owner(self):
  73         # A message posted to the list goes through the posting chain.
  74         msgdata = dict(listid='test.example.com',
  75                        to_owner=True)
  76         config.switchboards['in'].enqueue(self._msg, msgdata)
  77         self._in.run()
  78         items = get_queue_messages('out', expected_count=1)
  79         self.assertEqual(items[0].msgdata.get('marker'), 'owner')