src/postorius/tests/mailman_api_tests/test_subscriptions.py

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