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