src/mailman/rest/tests/test_moderation.py

   1 # Copyright (C) 2012-2015 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 """REST moderation tests."""
  19
  20 __all__ = [
  21     'TestPostModeration',
  22     'TestSubscriptionModeration',
  23     ]
  24
  25
  26 import unittest
  27
  28 from mailman.app.lifecycle import create_list
  29 from mailman.app.moderator import hold_message
  30 from mailman.database.transaction import transaction
  31 from mailman.interfaces.mailinglist import SubscriptionPolicy
  32 from mailman.interfaces.registrar import IRegistrar
  33 from mailman.interfaces.usermanager import IUserManager
  34 from mailman.testing.helpers import (
  35     call_api, get_queue_messages, specialized_message_from_string as mfs)
  36 from mailman.testing.layers import RESTLayer
  37 from mailman.utilities.datetime import now
  38 from urllib.error import HTTPError
  39 from zope.component import getUtility
  40
  41
  42 \f
  43 class TestPostModeration(unittest.TestCase):
  44     layer = RESTLayer
  45
  46     def setUp(self):
  47         with transaction():
  48             self._mlist = create_list('ant@example.com')
  49         self._msg = mfs("""\
  50 From: anne@example.com
  51 To: ant@example.com
  52 Subject: Something
  53 Message-ID: <alpha>
  54
  55 Something else.
  56 """)
  57
  58     def test_not_found(self):
  59         # When a bogus mailing list is given, 404 should result.
  60         with self.assertRaises(HTTPError) as cm:
  61             call_api('http://localhost:9001/3.0/lists/bee@example.com/held')
  62         self.assertEqual(cm.exception.code, 404)
  63
  64     def test_bad_held_message_request_id(self):
  65         # Bad request when request_id is not an integer.
  66         with self.assertRaises(HTTPError) as cm:
  67             call_api(
  68                 'http://localhost:9001/3.0/lists/ant@example.com/held/bogus')
  69         self.assertEqual(cm.exception.code, 400)
  70
  71     def test_missing_held_message_request_id(self):
  72         # Bad request when the request_id is not in the database.
  73         with self.assertRaises(HTTPError) as cm:
  74             call_api('http://localhost:9001/3.0/lists/ant@example.com/held/99')
  75         self.assertEqual(cm.exception.code, 404)
  76
  77     def test_bad_held_message_action(self):
  78         # POSTing to a held message with a bad action.
  79         held_id = hold_message(self._mlist, self._msg)
  80         url = 'http://localhost:9001/3.0/lists/ant@example.com/held/{0}'
  81         with self.assertRaises(HTTPError) as cm:
  82             call_api(url.format(held_id), {'action': 'bogus'})
  83         self.assertEqual(cm.exception.code, 400)
  84         self.assertEqual(cm.exception.msg,
  85                          b'Cannot convert parameters: action')
  86
  87     def test_discard(self):
  88         # Discarding a message removes it from the moderation queue.
  89         with transaction():
  90             held_id = hold_message(self._mlist, self._msg)
  91         url = 'http://localhost:9001/3.0/lists/ant@example.com/held/{}'.format(
  92             held_id)
  93         content, response = call_api(url, dict(action='discard'))
  94         self.assertEqual(response.status, 204)
  95         # Now it's gone.
  96         with self.assertRaises(HTTPError) as cm:
  97             call_api(url, dict(action='discard'))
  98         self.assertEqual(cm.exception.code, 404)
  99
 100
 101 \f
 102 class TestSubscriptionModeration(unittest.TestCase):
 103     layer = RESTLayer
 104     maxDiff = None
 105
 106     def setUp(self):
 107         with transaction():
 108             self._mlist = create_list('ant@example.com')
 109             self._registrar = IRegistrar(self._mlist)
 110             manager = getUtility(IUserManager)
 111             self._anne = manager.create_address(
 112                 'anne@example.com', 'Anne Person')
 113             self._bart = manager.make_user(
 114                 'bart@example.com', 'Bart Person')
 115             preferred = list(self._bart.addresses)[0]
 116             preferred.verified_on = now()
 117             self._bart.preferred_address = preferred
 118
 119     def test_no_such_list(self):
 120         # Try to get the requests of a nonexistent list.
 121         with self.assertRaises(HTTPError) as cm:
 122             call_api('http://localhost:9001/3.0/lists/bee@example.com/'
 123                      'requests')
 124         self.assertEqual(cm.exception.code, 404)
 125
 126     def test_no_such_subscription_token(self):
 127         # Bad request when the token is not in the database.
 128         with self.assertRaises(HTTPError) as cm:
 129             call_api('http://localhost:9001/3.0/lists/ant@example.com/'
 130                      'requests/missing')
 131         self.assertEqual(cm.exception.code, 404)
 132
 133     def test_bad_subscription_action(self):
 134         # POSTing to a held message with a bad action.
 135         token, token_owner, member = self._registrar.register(self._anne)
 136         # Anne's subscription request got held.
 137         self.assertIsNone(member)
 138         # Let's try to handle her request, but with a bogus action.
 139         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 140         with self.assertRaises(HTTPError) as cm:
 141             call_api(url.format(token), dict(
 142                 action='bogus',
 143                 ))
 144         self.assertEqual(cm.exception.code, 400)
 145         self.assertEqual(cm.exception.msg,
 146                          b'Cannot convert parameters: action')
 147
 148     def test_list_held_requests(self):
 149         # We can view all the held requests.
 150         with transaction():
 151             token_1, token_owner, member = self._registrar.register(self._anne)
 152             # Anne's subscription request got held.
 153             self.assertIsNotNone(token_1)
 154             self.assertIsNone(member)
 155             token_2, token_owner, member = self._registrar.register(self._bart)
 156             self.assertIsNotNone(token_2)
 157             self.assertIsNone(member)
 158         content, response = call_api(
 159             'http://localhost:9001/3.0/lists/ant@example.com/requests')
 160         self.assertEqual(response.status, 200)
 161         self.assertEqual(content['total_size'], 2)
 162         tokens = set(json['token'] for json in content['entries'])
 163         self.assertEqual(tokens, {token_1, token_2})
 164         emails = set(json['email'] for json in content['entries'])
 165         self.assertEqual(emails, {'anne@example.com', 'bart@example.com'})
 166
 167     def test_individual_request(self):
 168         # We can view an individual request.
 169         with transaction():
 170             token, token_owner, member = self._registrar.register(self._anne)
 171             # Anne's subscription request got held.
 172             self.assertIsNotNone(token)
 173             self.assertIsNone(member)
 174         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 175         content, response = call_api(url.format(token))
 176         self.assertEqual(response.status, 200)
 177         self.assertEqual(content['token'], token)
 178         self.assertEqual(content['token_owner'], token_owner.name)
 179         self.assertEqual(content['email'], 'anne@example.com')
 180
 181     def test_accept(self):
 182         # POST to the request to accept it.
 183         with transaction():
 184             token, token_owner, member = self._registrar.register(self._anne)
 185         # Anne's subscription request got held.
 186         self.assertIsNone(member)
 187         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 188         content, response = call_api(url.format(token), dict(
 189             action='accept',
 190             ))
 191         self.assertEqual(response.status, 204)
 192         # Anne is a member.
 193         self.assertEqual(
 194             self._mlist.members.get_member('anne@example.com').address,
 195             self._anne)
 196         # The request URL no longer exists.
 197         with self.assertRaises(HTTPError) as cm:
 198             call_api(url.format(token), dict(
 199                 action='accept',
 200                 ))
 201         self.assertEqual(cm.exception.code, 404)
 202
 203     def test_accept_bad_token(self):
 204         # Try to accept a request with a bogus token.
 205         with self.assertRaises(HTTPError) as cm:
 206             call_api('http://localhost:9001/3.0/lists/ant@example.com'
 207                      '/requests/bogus',
 208                      dict(action='accept'))
 209         self.assertEqual(cm.exception.code, 404)
 210
 211     def test_accept_by_moderator_clears_request_queue(self):
 212         # After accepting a message held for moderator approval, there are no
 213         # more requests to handle.
 214         #
 215         # We start with nothing in the queue.
 216         content, response = call_api(
 217             'http://localhost:9001/3.0/lists/ant@example.com/requests')
 218         self.assertEqual(content['total_size'], 0)
 219         # Anne tries to subscribe to a list that only requests moderator
 220         # approval.
 221         with transaction():
 222             self._mlist.subscription_policy = SubscriptionPolicy.moderate
 223             token, token_owner, member = self._registrar.register(
 224                 self._anne,
 225                 pre_verified=True, pre_confirmed=True)
 226         # There's now one request in the queue, and it's waiting on moderator
 227         # approval.
 228         content, response = call_api(
 229             'http://localhost:9001/3.0/lists/ant@example.com/requests')
 230         self.assertEqual(content['total_size'], 1)
 231         json = content['entries'][0]
 232         self.assertEqual(json['token_owner'], 'moderator')
 233         self.assertEqual(json['email'], 'anne@example.com')
 234         # The moderator approves the request.
 235         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 236         content, response = call_api(url.format(token), {'action': 'accept'})
 237         self.assertEqual(response.status, 204)
 238         # And now the request queue is empty.
 239         content, response = call_api(
 240             'http://localhost:9001/3.0/lists/ant@example.com/requests')
 241         self.assertEqual(content['total_size'], 0)
 242
 243     def test_discard(self):
 244         # POST to the request to discard it.
 245         with transaction():
 246             token, token_owner, member = self._registrar.register(self._anne)
 247         # Anne's subscription request got held.
 248         self.assertIsNone(member)
 249         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 250         content, response = call_api(url.format(token), dict(
 251             action='discard',
 252             ))
 253         self.assertEqual(response.status, 204)
 254         # Anne is not a member.
 255         self.assertIsNone(self._mlist.members.get_member('anne@example.com'))
 256         # The request URL no longer exists.
 257         with self.assertRaises(HTTPError) as cm:
 258             call_api(url.format(token), dict(
 259                 action='discard',
 260                 ))
 261         self.assertEqual(cm.exception.code, 404)
 262
 263     def test_defer(self):
 264         # Defer the decision for some other moderator.
 265         with transaction():
 266             token, token_owner, member = self._registrar.register(self._anne)
 267         # Anne's subscription request got held.
 268         self.assertIsNone(member)
 269         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 270         content, response = call_api(url.format(token), dict(
 271             action='defer',
 272             ))
 273         self.assertEqual(response.status, 204)
 274         # Anne is not a member.
 275         self.assertIsNone(self._mlist.members.get_member('anne@example.com'))
 276         # The request URL still exists.
 277         content, response = call_api(url.format(token), dict(
 278             action='defer',
 279             ))
 280         self.assertEqual(response.status, 204)
 281         # And now we can accept it.
 282         content, response = call_api(url.format(token), dict(
 283             action='accept',
 284             ))
 285         self.assertEqual(response.status, 204)
 286         # Anne is a member.
 287         self.assertEqual(
 288             self._mlist.members.get_member('anne@example.com').address,
 289             self._anne)
 290         # The request URL no longer exists.
 291         with self.assertRaises(HTTPError) as cm:
 292             call_api(url.format(token), dict(
 293                 action='accept',
 294                 ))
 295         self.assertEqual(cm.exception.code, 404)
 296
 297     def test_defer_bad_token(self):
 298         # Try to accept a request with a bogus token.
 299         with self.assertRaises(HTTPError) as cm:
 300             call_api('http://localhost:9001/3.0/lists/ant@example.com'
 301                      '/requests/bogus',
 302                      dict(action='defer'))
 303         self.assertEqual(cm.exception.code, 404)
 304
 305     def test_reject(self):
 306         # POST to the request to reject it.  This leaves a bounce message in
 307         # the virgin queue.
 308         with transaction():
 309             token, token_owner, member = self._registrar.register(self._anne)
 310         # Anne's subscription request got held.
 311         self.assertIsNone(member)
 312         # Clear out the virgin queue, which currently contains the
 313         # confirmation message sent to Anne.
 314         get_queue_messages('virgin')
 315         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 316         content, response = call_api(url.format(token), dict(
 317             action='reject',
 318             ))
 319         self.assertEqual(response.status, 204)
 320         # Anne is not a member.
 321         self.assertIsNone(self._mlist.members.get_member('anne@example.com'))
 322         # The request URL no longer exists.
 323         with self.assertRaises(HTTPError) as cm:
 324             call_api(url.format(token), dict(
 325                 action='reject',
 326                 ))
 327         self.assertEqual(cm.exception.code, 404)
 328         # And the rejection message to Anne is now in the virgin queue.
 329         items = get_queue_messages('virgin')
 330         self.assertEqual(len(items), 1)
 331         message = items[0].msg
 332         self.assertEqual(message['From'], 'ant-bounces@example.com')
 333         self.assertEqual(message['To'], 'anne@example.com')
 334         self.assertEqual(message['Subject'],
 335                          'Request to mailing list "Ant" rejected')
 336
 337     def test_reject_bad_token(self):
 338         # Try to accept a request with a bogus token.
 339         with self.assertRaises(HTTPError) as cm:
 340             call_api('http://localhost:9001/3.0/lists/ant@example.com'
 341                      '/requests/bogus',
 342                      dict(action='reject'))
 343         self.assertEqual(cm.exception.code, 404)