src/mailman/runners/tests/test_join.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 mailing list joins."""
  19
  20 __all__ = [
  21     'TestJoin',
  22     'TestJoinWithDigests',
  23     ]
  24
  25
  26 import unittest
  27
  28 from email.iterators import body_line_iterator
  29 from mailman.app.lifecycle import create_list
  30 from mailman.config import config
  31 from mailman.interfaces.member import DeliveryMode
  32 from mailman.interfaces.registrar import IRegistrar
  33 from mailman.interfaces.subscriptions import ISubscriptionService, TokenOwner
  34 from mailman.interfaces.usermanager import IUserManager
  35 from mailman.runners.command import CommandRunner
  36 from mailman.testing.helpers import (
  37     get_queue_messages, make_testable_runner, reset_the_world,
  38     specialized_message_from_string as mfs)
  39 from mailman.testing.layers import ConfigLayer
  40 from zope.component import getUtility
  41
  42
  43 \f
  44 class TestJoin(unittest.TestCase):
  45     """Test mailing list joins."""
  46
  47     layer = ConfigLayer
  48
  49     def setUp(self):
  50         self._mlist = create_list('test@example.com')
  51         self._mlist.send_welcome_message = False
  52         self._commandq = config.switchboards['command']
  53         self._runner = make_testable_runner(CommandRunner, 'command')
  54
  55     def tearDown(self):
  56         reset_the_world()
  57
  58     def test_double_confirmation(self):
  59         # A join request comes in using both the -join address and the word
  60         # 'subscribe' in the first line of the body.  This should produce just
  61         # one subscription request and one confirmation response.
  62         msg = mfs("""\
  63 From: anne@example.org
  64 To: test-join@example.com
  65
  66 subscribe
  67 """)
  68         # Adding the subaddress to the metadata dictionary mimics what happens
  69         # when the above email message is first processed by the lmtp runner.
  70         # For convenience, we skip that step in this test.
  71         self._commandq.enqueue(msg, dict(listid='test.example.com',
  72                                          subaddress='join'))
  73         self._runner.run()
  74         # There will be two messages in the queue.  The first one is a reply
  75         # to Anne notifying her of the status of her command email.  The
  76         # second one is the confirmation message of her join request.
  77         messages = get_queue_messages('virgin', sort_on='subject')
  78         self.assertEqual(len(messages), 2)
  79         self.assertTrue(str(messages[1].msg['subject']).startswith('confirm'))
  80         self.assertEqual(messages[0].msg['subject'],
  81                          'The results of your email commands')
  82         # Search the contents of the results message.  There should be just
  83         # one 'Confirmation email' line.
  84         confirmation_lines = []
  85         in_results = False
  86         for line in body_line_iterator(messages[0].msg):
  87             line = line.strip()
  88             if in_results:
  89                 if line.startswith('- Done'):
  90                     break
  91                 if len(line) > 0:
  92                     confirmation_lines.append(line)
  93             if line.strip() == '- Results:':
  94                 in_results = True
  95         # There should be exactly one confirmation line.
  96         self.assertEqual(len(confirmation_lines), 1)
  97         # And the confirmation line should name Anne's email address.
  98         self.assertTrue('anne@example.org' in confirmation_lines[0])
  99
 100     def test_join_when_already_a_member(self):
 101         anne = getUtility(IUserManager).create_user('anne@example.org')
 102         self._mlist.subscribe(list(anne.addresses)[0])
 103         # When someone tries to join by email and they are already a member,
 104         # ignore the request.
 105         msg = mfs("""\
 106 From: anne@example.org
 107 To: test-join@example.com
 108 Subject: join
 109
 110 """)
 111         self._commandq.enqueue(msg, dict(listid='test.example.com'))
 112         self._runner.run()
 113         # There will be one message in the queue - a reply to Anne notifying
 114         # her of the status of her command email.  Because Anne is already
 115         # subscribed to the list, she gets and needs no confirmation.
 116         messages = get_queue_messages('virgin')
 117         self.assertEqual(len(messages), 1)
 118         self.assertEqual(messages[0].msg['subject'],
 119                          'The results of your email commands')
 120         # Search the contents of the results message.  There should be just
 121         # one 'Confirmation email' line.
 122         confirmation_lines = []
 123         in_results = False
 124         for line in body_line_iterator(messages[0].msg):
 125             line = line.strip()
 126             if in_results:
 127                 if line.startswith('- Done'):
 128                     break
 129                 if len(line) > 0:
 130                     confirmation_lines.append(line)
 131             if line.strip() == '- Results:':
 132                 in_results = True
 133         # There should be exactly one confirmation line.
 134         self.assertEqual(len(confirmation_lines), 1)
 135         # And the confirmation line should name Anne's email address.
 136         self.assertTrue('anne@example.org' in confirmation_lines[0])
 137
 138
 139 \f
 140 class TestJoinWithDigests(unittest.TestCase):
 141     """Test mailing list joins with the digests=<no|mime|plain> argument."""
 142
 143     layer = ConfigLayer
 144
 145     def setUp(self):
 146         self._mlist = create_list('test@example.com')
 147         self._commandq = config.switchboards['command']
 148         self._runner = make_testable_runner(CommandRunner, 'command')
 149
 150     def tearDown(self):
 151         reset_the_world()
 152
 153     def _confirm(self):
 154         # There will be two messages in the queue - the confirmation messages,
 155         # and a reply to Anne notifying her of the status of her command
 156         # email.  We need to dig the confirmation token out of the Subject
 157         # header of the latter so that we can confirm the subscription.
 158         messages = get_queue_messages('virgin', sort_on='subject')
 159         self.assertEqual(len(messages), 2)
 160         subject_words = str(messages[1].msg['subject']).split()
 161         self.assertEqual(subject_words[0], 'confirm')
 162         token = subject_words[1]
 163         token, token_owner, rmember = IRegistrar(self._mlist).confirm(token)
 164         self.assertIsNone(token)
 165         self.assertEqual(token_owner, TokenOwner.no_one)
 166         # Now, make sure that Anne is a member of the list and is receiving
 167         # digest deliveries.
 168         members = getUtility(ISubscriptionService).find_members(
 169             'anne@example.org')
 170         self.assertEqual(len(members), 1)
 171         self.assertEqual(rmember, members[0])
 172         return rmember
 173
 174     def test_join_with_implicit_no_digests(self):
 175         # Test the digest=mime argument to the join command.
 176         msg = mfs("""\
 177 From: anne@example.org
 178 To: test-request@example.com
 179
 180 join
 181 """)
 182         self._commandq.enqueue(msg, dict(listid='test.example.com'))
 183         self._runner.run()
 184         anne = self._confirm()
 185         self.assertEqual(anne.address.email, 'anne@example.org')
 186         self.assertEqual(anne.delivery_mode, DeliveryMode.regular)
 187
 188     def test_join_with_explicit_no_digests(self):
 189         # Test the digest=mime argument to the join command.
 190         msg = mfs("""\
 191 From: anne@example.org
 192 To: test-request@example.com
 193
 194 join digest=no
 195 """)
 196         self._commandq.enqueue(msg, dict(listid='test.example.com'))
 197         self._runner.run()
 198         anne = self._confirm()
 199         self.assertEqual(anne.address.email, 'anne@example.org')
 200         self.assertEqual(anne.delivery_mode, DeliveryMode.regular)
 201
 202     # LP: #1444184 - digest=mime is not currently supported.
 203     @unittest.expectedFailure
 204     def test_join_with_mime_digests(self):
 205         # Test the digest=mime argument to the join command.
 206         msg = mfs("""\
 207 From: anne@example.org
 208 To: test-request@example.com
 209
 210 join digest=mime
 211 """)
 212         self._commandq.enqueue(msg, dict(listid='test.example.com'))
 213         self._runner.run()
 214         anne = self._confirm()
 215         self.assertEqual(anne.address.email, 'anne@example.org')
 216         self.assertEqual(anne.delivery_mode, DeliveryMode.mime_digests)
 217
 218     # LP: #1444184 - digest=mime is not currently supported.
 219     @unittest.expectedFailure
 220     def test_join_with_plain_digests(self):
 221         # Test the digest=mime argument to the join command.
 222         msg = mfs("""\
 223 From: anne@example.org
 224 To: test-request@example.com
 225
 226 join digest=plain
 227 """)
 228         self._commandq.enqueue(msg, dict(listid='test.example.com'))
 229         self._runner.run()
 230         anne = self._confirm()
 231         self.assertEqual(anne.address.email, 'anne@example.org')
 232         self.assertEqual(anne.delivery_mode, DeliveryMode.plaintext_digests)