src/mailman/commands/tests/test_eml_membership.py

   1 # Copyright (C) 2016-2023 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 <https://www.gnu.org/licenses/>.
  17
  18 """Test the Join and Leave commands."""
  19
  20 import unittest
  21
  22 from mailman.app.lifecycle import create_list
  23 from mailman.commands.eml_membership import Join, Leave
  24 from mailman.email.message import Message
  25 from mailman.interfaces.bans import IBanManager
  26 from mailman.interfaces.mailinglist import SubscriptionPolicy
  27 from mailman.interfaces.member import DeliveryMode
  28 from mailman.interfaces.pending import IPendings
  29 from mailman.interfaces.subscriptions import ISubscriptionManager
  30 from mailman.interfaces.usermanager import IUserManager
  31 from mailman.runners.command import Results
  32 from mailman.testing.helpers import set_preferred
  33 from mailman.testing.layers import ConfigLayer
  34 from zope.component import getUtility
  35
  36
  37 class TestLeave(unittest.TestCase):
  38     layer = ConfigLayer
  39
  40     def setUp(self):
  41         self._mlist = create_list('ant@example.com')
  42         self._command = Leave()
  43
  44     def test_confirm_leave_not_a_member(self):
  45         self._mlist.unsubscription_policy = SubscriptionPolicy.confirm
  46         # Try to unsubscribe someone who is not a member.  Anne is a real
  47         # user, with a validated address, but she is not a member of the
  48         # mailing list.
  49         anne = getUtility(IUserManager).create_user('anne@example.com')
  50         set_preferred(anne)
  51         # Initiate an unsubscription.
  52         msg = Message()
  53         msg['From'] = 'anne@example.com'
  54         results = Results()
  55         self._command.process(self._mlist, msg, {}, (), results)
  56         self.assertEqual(
  57             str(results).splitlines()[-1],
  58             'leave: anne@example.com is not a member of ant@example.com')
  59
  60     def test_leave_no_sender(self):
  61         # Initiate an unsubscription with no sender in the message.
  62         msg = Message()
  63         results = Results()
  64         self._command.process(self._mlist, msg, {}, (), results)
  65         self.assertEqual(
  66             str(results).splitlines()[-1],
  67             'leave: No valid email address found to unsubscribe')
  68
  69     def test_leave_no_user(self):
  70         # Initiate an unsubscription for a non-user.
  71         msg = Message()
  72         msg['From'] = 'anne@example.com'
  73         results = Results()
  74         self._command.process(self._mlist, msg, {}, (), results)
  75         self.assertEqual(
  76             str(results).splitlines()[-1],
  77             'No registered user for email address: anne@example.com')
  78
  79
  80 class TestJoin(unittest.TestCase):
  81     layer = ConfigLayer
  82
  83     def setUp(self):
  84         self._mlist = create_list('ant@example.com')
  85         self._command = Join()
  86
  87     def test_join_successful(self):
  88         # Subscribe a member via join.
  89         msg = Message()
  90         msg['From'] = 'anne@example.com'
  91         results = Results()
  92         self._command.process(self._mlist, msg, {}, (), results)
  93         self.assertIn('Confirmation email sent to anne@example.com',
  94                       str(results))
  95
  96     def test_join_rfc2047_display(self):
  97         # Subscribe a member with RFC 2047 encoded display name via join.
  98         msg = Message()
  99         msg['From'] = '=?utf-8?q?Anne?= <anne@example.com>'
 100         results = Results()
 101         self._command.process(self._mlist, msg, {}, (), results)
 102         self.assertIn('Confirmation email sent to Anne <anne@example.com>',
 103                       str(results))
 104         # Check the pending confirmation.
 105         pendings = list(getUtility(IPendings).find(self._mlist,
 106                                                    'subscription',
 107                                                    confirm=False))
 108         self.assertEqual(1, len(pendings))
 109         token = pendings[0][0]
 110         pended = getUtility(IPendings).confirm(token, expunge=False)
 111         self.assertEqual('Anne', pended['display_name'])
 112         self.assertEqual('anne@example.com', pended['email'])
 113
 114     def test_join_rfc2047_display_with_comma(self):
 115         # Subscribe a member with RFC 2047 encoded display name containing a
 116         # comma and non-ascii via join.
 117         msg = Message()
 118         msg['From'] = '=?utf-8?q?J=C3=BCnk=2C_Anne?= <anne@example.com>'
 119         results = Results()
 120         self._command.process(self._mlist, msg, {}, (), results)
 121         self.assertIn('Confirmation email sent to =?utf-8?b?SsO8bmssIEFubmU=?='
 122                       ' <anne@example.com>', str(results))
 123         # Check the pending confirmation.
 124         pendings = list(getUtility(IPendings).find(self._mlist,
 125                                                    'subscription',
 126                                                    confirm=False))
 127         self.assertEqual(1, len(pendings))
 128         token = pendings[0][0]
 129         pended = getUtility(IPendings).confirm(token, expunge=False)
 130         self.assertEqual('Jünk, Anne', pended['display_name'])
 131         self.assertEqual('anne@example.com', pended['email'])
 132
 133     def test_join_digest(self):
 134         # Subscribe a member to digest via join.
 135         msg = Message()
 136         msg['From'] = 'anne@example.com'
 137         results = Results()
 138         self._command.process(self._mlist, msg, {}, ('digest=mime',), results)
 139         self.assertIn('Confirmation email sent to anne@example.com',
 140                       str(results))
 141
 142     def test_join_digest_works(self):
 143         # Subscribe a member to digest via join and verify it works.
 144         # Set things so the subscribe works now.
 145         self._mlist.subscription_policy = SubscriptionPolicy.open
 146         # Create a verified address for Anne.
 147         user = getUtility(IUserManager).make_user('anne@example.com', 'Anne')
 148         address = user.addresses[0]
 149         address.verified_on = address.registered_on
 150         msg = Message()
 151         msg['From'] = 'anne@example.com'
 152         results = Results()
 153         self._command.process(self._mlist, msg, {}, ('digest=mime',), results)
 154         # Anne is a member.
 155         members = list(self._mlist.members.members)
 156         self.assertEqual(1, len(members))
 157         self.assertEqual('anne@example.com', members[0].address.email)
 158         self.assertEqual(DeliveryMode.mime_digests, members[0].delivery_mode)
 159
 160     def test_join_other(self):
 161         # Subscribe a different address via join.
 162         msg = Message()
 163         msg['From'] = 'anne@example.com'
 164         results = Results()
 165         self._command.process(self._mlist, msg, {},
 166                               ('address=bob@example.com',), results)
 167         self.assertIn('Confirmation email sent to bob@example.com',
 168                       str(results))
 169
 170     def test_join_other_bogus(self):
 171         # Try to subscribe a bogus different address via join.
 172         msg = Message()
 173         msg['From'] = 'anne@example.com'
 174         results = Results()
 175         self._command.process(self._mlist, msg, {},
 176                               ('address=bogus',), results)
 177         self.assertIn('Invalid email address: bogus', str(results))
 178
 179     def test_join_bad_argument(self):
 180         # Try to subscribe a member with a bad argument via join.
 181         msg = Message()
 182         msg['From'] = 'anne@example.com'
 183         results = Results()
 184         self._command.process(self._mlist, msg, {}, ('digest=bogus',), results)
 185         self.assertIn('bad argument: digest=bogus', str(results))
 186
 187     def test_join_bad_argument_name(self):
 188         # Try to subscribe a member with a bad argument via join.
 189         msg = Message()
 190         msg['From'] = 'anne@example.com'
 191         results = Results()
 192         self._command.process(self._mlist, msg, {}, ('reg=bogus',), results)
 193         self.assertIn('bad argument: reg=bogus', str(results))
 194
 195     def test_join_bad_argument_no_equal(self):
 196         # Try to subscribe a member with a bad argument via join.
 197         msg = Message()
 198         msg['From'] = 'anne@example.com'
 199         results = Results()
 200         self._command.process(self._mlist, msg, {}, ('digest',), results)
 201         self.assertIn('bad argument: digest', str(results))
 202
 203     def test_join_already_a_member(self):
 204         # Try to subscribe someone who is already a member.  Anne is a real
 205         # user, with a validated address, but she is not a member of the
 206         # mailing list yet.
 207         anne = getUtility(IUserManager).create_user('anne@example.com')
 208         set_preferred(anne)
 209         # First subscribe anne.
 210         ISubscriptionManager(self._mlist).register(anne, pre_verified=True,
 211                                                    pre_confirmed=True,
 212                                                    pre_approved=True)
 213         # Then initiate a subscription.
 214         msg = Message()
 215         msg['From'] = 'anne@example.com'
 216         results = Results()
 217         self._command.process(self._mlist, msg, {}, (), results)
 218         self.assertEqual(
 219             str(results).splitlines()[-1],
 220             'anne@example.com is already a member of '
 221             'mailing list ant@example.com')
 222
 223     def test_join_banned(self):
 224         # Try to subscribe someone who is banned.  Anne is a real
 225         # user, with a validated address, but she is not a member of the
 226         # mailing list and is banned from joining.
 227         # Add anne to the ban list.
 228         IBanManager(self._mlist).ban('anne@example.com')
 229         # Then initiate a subscription.
 230         msg = Message()
 231         msg['From'] = 'anne@example.com'
 232         results = Results()
 233         self._command.process(self._mlist, msg, {}, (), results)
 234         self.assertEqual(
 235             str(results).splitlines()[-1],
 236             'anne@example.com is not allowed to subscribe to ant@example.com')
 237
 238     def test_join_pending(self):
 239         self._mlist.subscription_policy = SubscriptionPolicy.confirm
 240         # Try to subscribe someone who already has a subscription pending.
 241         # Anne is a real user, with a validated address, who already has a
 242         # pending subscription for this mailing list.
 243         anne = getUtility(IUserManager).create_user('anne@example.com')
 244         set_preferred(anne)
 245         # Initiate a subscription.
 246         ISubscriptionManager(self._mlist).register(anne)
 247         # And try to subscribe.
 248         msg = Message()
 249         msg['From'] = 'anne@example.com'
 250         results = Results()
 251         self._command.process(self._mlist, msg, {}, (), results)
 252         self.assertEqual(
 253             str(results).splitlines()[-1],
 254             'anne@example.com has a pending subscription for ant@example.com')
 255
 256     def test_join_posting_address(self):
 257         # Try to subscribe the list posting address.
 258         msg = Message()
 259         msg['From'] = self._mlist.posting_address
 260         results = Results()
 261         self._command.process(self._mlist, msg, {}, (), results)
 262         self.assertEqual(
 263             str(results).splitlines()[-1],
 264             'List posting address not allowed')