src/postorius/tests/mailman_api_tests/test_subscriptions.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright (C) 2012-2023 by the Free Software Foundation, Inc.
   3 #
   4 # This file is part of Postorius.
   5 #
   6 # Postorius is free software: you can redistribute it and/or modify it under
   7 # the terms of the GNU General Public License as published by the Free
   8 # Software Foundation, either version 3 of the License, or (at your option)
   9 # any later version.
  10 # Postorius 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 # Postorius.  If not, see <http://www.gnu.org/licenses/>.
  17
  18
  19 from unittest import expectedFailure
  20 from unittest.mock import patch
  21
  22 from django.contrib.auth.models import User
  23 from django.urls import reverse
  24
  25 from allauth.account.models import EmailAddress
  26 from django_mailman3.lib.mailman import get_mailman_user
  27
  28 from postorius.models import SubscriptionMode
  29 from postorius.tests.utils import ViewTestCase
  30
  31
  32 class TestSubscription(ViewTestCase):
  33     """Tests subscription to lists"""
  34
  35     def setUp(self):
  36         super(TestSubscription, self).setUp()
  37         self.domain = self.mm_client.create_domain('example.com')
  38         self.open_list = self.domain.create_list('open_list')
  39         # Set subscription policy to open
  40         settings = self.open_list.settings
  41         settings['subscription_policy'] = 'open'
  42         settings.save()
  43         self.mod_list = self.domain.create_list('moderate_subs')
  44         # Set subscription policy to moderate
  45         settings = self.mod_list.settings
  46         settings['subscription_policy'] = 'moderate'
  47         settings.save()
  48         # Create django user.
  49         self.user = User.objects.create_user(
  50             'testuser', 'test@example.com', 'pwd'
  51         )
  52         EmailAddress.objects.create(
  53             user=self.user, email=self.user.email, verified=True
  54         )
  55         EmailAddress.objects.create(
  56             user=self.user, email='fritz@example.org', verified=True
  57         )
  58         # Create Mailman user
  59         self.mm_user = self.mm_client.create_user('test@example.com', '')
  60         self.mm_user.add_address('fritz@example.org')
  61         for address in self.mm_user.addresses:
  62             address.verify()
  63
  64     @patch('mailmanclient.MailingList.subscribe')
  65     def test_anonymous_subscribe(self, mock_subscribe):
  66         response = self.client.post(
  67             reverse(
  68                 'list_anonymous_subscribe', args=('open_list.example.com',)
  69             ),
  70             {'email': 'test@example.com', 'display_name': 'Test User'},
  71         )
  72         mock_subscribe.assert_called_once()
  73         mock_subscribe.assert_called_with(
  74             'test@example.com',
  75             'Test User',
  76             pre_verified=False,
  77             pre_confirmed=False,
  78         )
  79         self.assertRedirects(
  80             response, reverse('list_summary', args=('open_list.example.com',))
  81         )
  82         self.assertHasSuccessMessage(response)
  83
  84     def test_subscribe_open(self):
  85         # The subscription goes straight through.
  86         self.client.login(username='testuser', password='pwd')
  87         response = self.client.post(
  88             reverse('list_subscribe', args=('open_list.example.com',)),
  89             {'subscriber': 'test@example.com'},
  90         )
  91         self.assertEqual(len(self.open_list.members), 1)
  92         self.assertEqual(len(self.open_list.requests), 0)
  93         self.assertRedirects(
  94             response, reverse('list_summary', args=('open_list.example.com',))
  95         )
  96         self.assertHasSuccessMessage(response)
  97
  98     def test_secondary_open(self):
  99         # Subscribe with a secondary email address
 100         self.client.login(username='testuser', password='pwd')
 101         response = self.client.post(
 102             reverse('list_subscribe', args=('open_list.example.com',)),
 103             {'subscriber': 'fritz@example.org'},
 104         )
 105         self.assertEqual(len(self.open_list.members), 1)
 106         self.assertEqual(len(self.open_list.requests), 0)
 107         self.assertRedirects(
 108             response, reverse('list_summary', args=('open_list.example.com',))
 109         )
 110         self.assertHasSuccessMessage(response)
 111
 112     def test_unknown_address(self):
 113         # Impossible to register with an unknown address
 114         self.client.login(username='testuser', password='pwd')
 115         response = self.client.post(
 116             reverse('list_subscribe', args=('open_list.example.com',)),
 117             {'subscriber': 'unknown@example.org'},
 118         )
 119         self.assertEqual(len(self.open_list.members), 0)
 120         self.assertEqual(len(self.open_list.requests), 0)
 121         self.assertRedirects(
 122             response, reverse('list_summary', args=('open_list.example.com',))
 123         )
 124         self.assertHasErrorMessage(response)
 125
 126     def test_banned_address(self):
 127         # Impossible to register with a banned address
 128         self.client.login(username='testuser', password='pwd')
 129         self.open_list.bans.add('test@example.com')
 130         response = self.client.post(
 131             reverse('list_subscribe', args=('open_list.example.com',)),
 132             {'subscriber': 'test@example.com'},
 133         )
 134         self.assertEqual(len(self.open_list.members), 0)
 135         self.assertEqual(len(self.open_list.requests), 0)
 136         self.assertRedirects(
 137             response, reverse('list_summary', args=('open_list.example.com',))
 138         )
 139         self.assertHasErrorMessage(response)
 140
 141     def test_subscribe_mod(self):
 142         # The subscription is held for approval.
 143         self.client.login(username='testuser', password='pwd')
 144         response = self.client.post(
 145             reverse('list_subscribe', args=('moderate_subs.example.com',)),
 146             {'subscriber': 'test@example.com'},
 147         )
 148         self.assertEqual(len(self.mod_list.members), 0)
 149         self.assertEqual(len(self.mod_list.requests), 1)
 150         self.assertRedirects(
 151             response,
 152             reverse('list_summary', args=('moderate_subs.example.com',)),
 153         )
 154         self.assertHasSuccessMessage(response)
 155
 156     def test_secondary_mod(self):
 157         # Subscribe with a secondary email address
 158         self.client.login(username='testuser', password='pwd')
 159         response = self.client.post(
 160             reverse('list_subscribe', args=('moderate_subs.example.com',)),
 161             {'subscriber': 'fritz@example.org'},
 162         )
 163         self.assertEqual(len(self.mod_list.members), 0)
 164         self.assertEqual(len(self.mod_list.requests), 1)
 165         self.assertRedirects(
 166             response,
 167             reverse('list_summary', args=('moderate_subs.example.com',)),
 168         )
 169         self.assertHasSuccessMessage(response)
 170
 171     def test_subscribe_already_pending(self):
 172         # The user tries to subscribe twice on a moderated list.
 173         self.client.login(username='testuser', password='pwd')
 174         response = self.client.post(
 175             reverse('list_subscribe', args=('moderate_subs.example.com',)),
 176             {'subscriber': 'test@example.com'},
 177         )
 178         self.assertEqual(len(self.mod_list.members), 0)
 179         self.assertEqual(len(self.mod_list.requests), 1)
 180         self.assertHasSuccessMessage(response)
 181         # Try to subscribe a second time.
 182         response = self.client.post(
 183             reverse('list_subscribe', args=('moderate_subs.example.com',)),
 184             {'subscriber': 'test@example.com'},
 185         )
 186         self.assertEqual(len(self.mod_list.members), 0)
 187         self.assertEqual(len(self.mod_list.requests), 1)
 188         message = self.assertHasErrorMessage(response)
 189         self.assertIn('Subscription request already pending', message)
 190
 191     def test_subscribe_with_name(self):
 192         owner = User.objects.create_user(
 193             'testowner', 'owner@example.com', 'pwd'
 194         )
 195         EmailAddress.objects.create(
 196             user=owner, email=owner.email, verified=True
 197         )
 198         self.open_list.add_owner('owner@example.com')
 199         self.client.login(username='testowner', password='pwd')
 200         email_list = """First Person <test-1@example.org>\n
 201                         "Second Person" <test-2@example.org>\n
 202                         test-3@example.org (Third Person)\n
 203                         test-4@example.org\n
 204                         <test-5@example.org>\n"""
 205         self.client.post(
 206             reverse('mass_subscribe', args=('open_list.example.com',)),
 207             {
 208                 'emails': email_list,
 209                 'pre_verified': True,
 210                 'send_welcome_message': 'default',
 211             },
 212         )
 213         self.assertEqual(len(self.open_list.members), 5)
 214         first = self.open_list.get_member('test-1@example.org')
 215         second = self.open_list.get_member('test-2@example.org')
 216         third = self.open_list.get_member('test-3@example.org')
 217         fourth = self.open_list.get_member('test-4@example.org')
 218         fifth = self.open_list.get_member('test-5@example.org')
 219         self.assertEqual(first.address.display_name, 'First Person')
 220         self.assertEqual(second.address.display_name, 'Second Person')
 221         self.assertEqual(third.address.display_name, 'Third Person')
 222         self.assertIsNone(fourth.address.display_name)
 223         self.assertIsNone(fifth.address.display_name)
 224
 225     def test_mass_subscribe_send_welcome_message(self):
 226         owner = User.objects.create_user(
 227             'testowner', 'owner@example.com', 'pwd'
 228         )
 229         EmailAddress.objects.create(
 230             user=owner, email=owner.email, verified=True
 231         )
 232         self.open_list.add_owner('owner@example.com')
 233         self.client.login(username='testowner', password='pwd')
 234         virgin_q = self.mm_client.queues['virgin']
 235         initial_files = len(virgin_q.files)
 236         email_list = """First Person <test-1@example.org>\n
 237                         "Second Person" <test-2@example.org>\n"""
 238         self.client.post(
 239             reverse('mass_subscribe', args=('open_list.example.com',)),
 240             {
 241                 'emails': email_list,
 242                 'pre_verified': True,
 243                 'send_welcome_message': True,
 244             },
 245         )
 246         self.assertEqual(len(self.open_list.members), 2)
 247         virgin_q = self.mm_client.queues['virgin']
 248         # There should be two more files in the queue.
 249         self.assertEqual(len(virgin_q.files) - initial_files, 2)
 250         initial_files += 2
 251         # Now subscribe some users by changing it to No.
 252         email_list = """test-3@example.org (Third Person)\n"""
 253         self.client.post(
 254             reverse('mass_subscribe', args=('open_list.example.com',)),
 255             {
 256                 'emails': email_list,
 257                 'pre_verified': True,
 258                 'send_welcome_message': False,
 259             },
 260         )
 261         self.assertEqual(len(self.open_list.members), 3)
 262         # There should be zero more messages in virgin queue because we set
 263         # `sent_welcome_message` to False.
 264         # There should be two more files in the queue.
 265         virgin_q = self.mm_client.queues['virgin']
 266         self.assertEqual(len(virgin_q.files) - initial_files, 0)
 267         # If set to none, the default value of send_welcome_message works,
 268         # which is True.
 269         email_list = """test4@example.org (Third Person)\n"""
 270         self.client.post(
 271             reverse('mass_subscribe', args=('open_list.example.com',)),
 272             {
 273                 'emails': email_list,
 274                 'pre_verified': True,
 275                 'send_welcome_message': 'default',
 276             },
 277         )
 278         self.assertEqual(len(self.open_list.members), 4)
 279         virgin_q = self.mm_client.queues['virgin']
 280         self.assertEqual(len(virgin_q.files) - initial_files, 1)
 281
 282     def test_change_subscription_open(self):
 283         # The subscription is changed from an address to another
 284         member = self.open_list.subscribe('test@example.com')
 285         self.assertEqual(len(self.open_list.members), 1)
 286         self.assertEqual(len(self.open_list.requests), 0)
 287         self.client.login(username='testuser', password='pwd')
 288         response = self.client.post(
 289             reverse('change_subscription', args=['open_list.example.com']),
 290             {'subscriber': 'fritz@example.org', 'member_id': member.member_id},
 291         )
 292         self.assertHasSuccessMessage(response)
 293         self.assertEqual(len(self.open_list.members), 1)
 294         self.assertEqual(len(self.open_list.requests), 0)
 295         try:
 296             member = self.open_list.get_member('fritz@example.org')
 297         except ValueError:
 298             self.fail('The subscription was not changed')
 299         self.assertEqual(member.email, 'fritz@example.org')
 300         self.assertRedirects(
 301             response, reverse('list_summary', args=('open_list.example.com',))
 302         )
 303
 304     def test_change_subscription_confirm(self):
 305         # The subscription is changed from an address to another
 306         confirm_list = self.domain.create_list('confirm_list')
 307         settings = confirm_list.settings
 308         settings['subscription_policy'] = 'confirm'
 309         settings.save()
 310         member = confirm_list.subscribe('test@example.com', pre_confirmed=True)
 311         self.assertEqual(len(confirm_list.members), 1)
 312         self.assertEqual(len(confirm_list.requests), 0)
 313         self.client.login(username='testuser', password='pwd')
 314         response = self.client.post(
 315             reverse('change_subscription', args=['confirm_list.example.com']),
 316             {'subscriber': 'fritz@example.org', 'member_id': member.member_id},
 317         )
 318         self.assertHasSuccessMessage(response)
 319         self.assertEqual(len(confirm_list.members), 1)
 320         self.assertEqual(len(confirm_list.requests), 0)
 321         try:
 322             member = confirm_list.get_member('fritz@example.org')
 323         except ValueError:
 324             self.fail('The subscription was not changed')
 325         self.assertEqual(member.email, 'fritz@example.org')
 326         self.assertRedirects(
 327             response,
 328             reverse('list_summary', args=('confirm_list.example.com',)),
 329         )
 330
 331     @expectedFailure
 332     def test_change_subscription_from_address_to_primary(self):
 333         # Test that can we can switch subscription between address and primary
 334         # address (which happens to be same).
 335         # This test is now expected to fail due to
 336         # https://gitlab.com/mailman/mailman/-/merge_requests/997 which no
 337         # longer permits a User whose primary address is the same as a
 338         # subscribed Address to subscribe and vice versa.
 339         member = self.open_list.subscribe('test@example.com')
 340         self.assertEqual(len(self.open_list.members), 1)
 341         self.assertEqual(
 342             member.subscription_mode, SubscriptionMode.as_address.name
 343         )
 344         self.assertEqual(len(self.open_list.requests), 0)
 345         self.client.login(username='testuser', password='pwd')
 346         mm_user = get_mailman_user(self.user)
 347         mm_user.preferred_address = 'test@example.com'
 348         # Switch subscription to primary address when the same address is
 349         # subscribed.
 350         response = self.client.post(
 351             reverse('change_subscription', args=['open_list.example.com']),
 352             {'subscriber': mm_user.user_id, 'member_id': member.member_id},
 353         )
 354         self.assertHasSuccessMessage(response)
 355         self.assertEqual(len(self.open_list.members), 1)
 356         self.assertEqual(len(self.open_list.requests), 0)
 357         try:
 358             member = self.open_list.get_member('test@example.com')
 359         except ValueError:
 360             self.fail('The subscription was not changed')
 361         self.assertEqual(
 362             member.subscription_mode, SubscriptionMode.as_user.name
 363         )
 364         self.assertRedirects(
 365             response, reverse('list_summary', args=('open_list.example.com',))
 366         )
 367         # Now, if the user_id and address both are subscribed.
 368         addr_member = self.open_list.subscribe('test@example.com')
 369         self.assertIsNotNone(addr_member)
 370         self.assertEqual(len(self.open_list.members), 2)
 371         # Now trying to switch subscription should say already subscribed.
 372         response = self.client.post(
 373             reverse('change_subscription', args=['open_list.example.com']),
 374             {
 375                 'subscriber': mm_user.user_id,
 376                 'member_id': addr_member.member_id,
 377             },
 378         )
 379         self.assertEqual(response.status_code, 302)
 380         self.assertHasErrorMessage(response)