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.registrar import IRegistrar
  32 from mailman.interfaces.usermanager import IUserManager
  33 from mailman.testing.helpers import (
  34     call_api, get_queue_messages, specialized_message_from_string as mfs)
  35 from mailman.testing.layers import RESTLayer
  36 from mailman.utilities.datetime import now
  37 from urllib.error import HTTPError
  38 from zope.component import getUtility
  39
  40
  41 \f
  42 class TestPostModeration(unittest.TestCase):
  43     layer = RESTLayer
  44
  45     def setUp(self):
  46         with transaction():
  47             self._mlist = create_list('ant@example.com')
  48         self._msg = mfs("""\
  49 From: anne@example.com
  50 To: ant@example.com
  51 Subject: Something
  52 Message-ID: <alpha>
  53
  54 Something else.
  55 """)
  56
  57     def test_not_found(self):
  58         # When a bogus mailing list is given, 404 should result.
  59         with self.assertRaises(HTTPError) as cm:
  60             call_api('http://localhost:9001/3.0/lists/bee@example.com/held')
  61         self.assertEqual(cm.exception.code, 404)
  62
  63     def test_bad_held_message_request_id(self):
  64         # Bad request when request_id is not an integer.
  65         with self.assertRaises(HTTPError) as cm:
  66             call_api(
  67                 'http://localhost:9001/3.0/lists/ant@example.com/held/bogus')
  68         self.assertEqual(cm.exception.code, 400)
  69
  70     def test_missing_held_message_request_id(self):
  71         # Bad request when the request_id is not in the database.
  72         with self.assertRaises(HTTPError) as cm:
  73             call_api('http://localhost:9001/3.0/lists/ant@example.com/held/99')
  74         self.assertEqual(cm.exception.code, 404)
  75
  76     def test_bad_held_message_action(self):
  77         # POSTing to a held message with a bad action.
  78         held_id = hold_message(self._mlist, self._msg)
  79         url = 'http://localhost:9001/3.0/lists/ant@example.com/held/{0}'
  80         with self.assertRaises(HTTPError) as cm:
  81             call_api(url.format(held_id), {'action': 'bogus'})
  82         self.assertEqual(cm.exception.code, 400)
  83         self.assertEqual(cm.exception.msg,
  84                          b'Cannot convert parameters: action')
  85
  86     def test_discard(self):
  87         # Discarding a message removes it from the moderation queue.
  88         with transaction():
  89             held_id = hold_message(self._mlist, self._msg)
  90         url = 'http://localhost:9001/3.0/lists/ant@example.com/held/{}'.format(
  91             held_id)
  92         content, response = call_api(url, dict(action='discard'))
  93         self.assertEqual(response.status, 204)
  94         # Now it's gone.
  95         with self.assertRaises(HTTPError) as cm:
  96             call_api(url, dict(action='discard'))
  97         self.assertEqual(cm.exception.code, 404)
  98
  99
 100 \f
 101 class TestSubscriptionModeration(unittest.TestCase):
 102     layer = RESTLayer
 103
 104     def setUp(self):
 105         with transaction():
 106             self._mlist = create_list('ant@example.com')
 107             self._registrar = IRegistrar(self._mlist)
 108             manager = getUtility(IUserManager)
 109             self._anne = manager.create_address(
 110                 'anne@example.com', 'Anne Person')
 111             self._bart = manager.make_user(
 112                 'bart@example.com', 'Bart Person')
 113             preferred = list(self._bart.addresses)[0]
 114             preferred.verified_on = now()
 115             self._bart.preferred_address = preferred
 116
 117     def test_no_such_list(self):
 118         # Try to get the requests of a nonexistent list.
 119         with self.assertRaises(HTTPError) as cm:
 120             call_api('http://localhost:9001/3.0/lists/bee@example.com/'
 121                      'requests')
 122         self.assertEqual(cm.exception.code, 404)
 123
 124     def test_no_such_subscription_token(self):
 125         # Bad request when the token is not in the database.
 126         with self.assertRaises(HTTPError) as cm:
 127             call_api('http://localhost:9001/3.0/lists/ant@example.com/'
 128                      'requests/missing')
 129         self.assertEqual(cm.exception.code, 404)
 130
 131     def test_bad_subscription_action(self):
 132         # POSTing to a held message with a bad action.
 133         token, token_owner, member = self._registrar.register(self._anne)
 134         # Anne's subscription request got held.
 135         self.assertIsNone(member)
 136         # Let's try to handle her request, but with a bogus action.
 137         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 138         with self.assertRaises(HTTPError) as cm:
 139             call_api(url.format(token), dict(
 140                 action='bogus',
 141                 ))
 142         self.assertEqual(cm.exception.code, 400)
 143         self.assertEqual(cm.exception.msg,
 144                          b'Cannot convert parameters: action')
 145
 146     def test_list_held_requests(self):
 147         # We can view all the held requests.
 148         token_1, token_owner, member = self._registrar.register(self._anne)
 149         # Anne's subscription request got held.
 150         self.assertIsNone(member)
 151         token_2, token_owner, member = self._registrar.register(self._bart)
 152         content, response = call_api(
 153             'http://localhost:9001/3.0/lists/ant@example.com/requests')
 154         self.assertEqual(response.status, 200)
 155         self.assertEqual(content['total_size'], 2)
 156         import pdb; pdb.set_trace()
 157
 158     def test_accept(self):
 159         # POST to the request to accept it.
 160         token, token_owner, member = self._registrar.register(self._anne)
 161         # Anne's subscription request got held.
 162         self.assertIsNone(member)
 163         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 164         with self.assertRaises(HTTPError) as cm:
 165             call_api(url.format(token), dict(
 166                 action='accept',
 167                 ))
 168         self.assertEqual(cm.exception.code, 204)
 169         # Anne is a member.
 170         self.assertEqual(
 171             self._mlist.get_members('anne@example.com').address,
 172             self._anne)
 173         # The request URL no longer exists.
 174         with self.assertRaises(HTTPError) as cm:
 175             call_api(url.format(token), dict(
 176                 action='accept',
 177                 ))
 178         self.assertEqual(cm.exception.code, 404)
 179
 180     def test_discard(self):
 181         # POST to the request to discard it.
 182         token, token_owner, member = self._registrar.register(self._anne)
 183         # Anne's subscription request got held.
 184         self.assertIsNone(member)
 185         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 186         with self.assertRaises(HTTPError) as cm:
 187             call_api(url.format(token), dict(
 188                 action='discard',
 189                 ))
 190         self.assertEqual(cm.exception.code, 204)
 191         # Anne is not a member.
 192         self.assertIsNone(self._mlist.get_members('anne@example.com'))
 193         # The request URL no longer exists.
 194         with self.assertRaises(HTTPError) as cm:
 195             call_api(url.format(token), dict(
 196                 action='discard',
 197                 ))
 198         self.assertEqual(cm.exception.code, 404)
 199
 200     def test_defer(self):
 201         # Defer the decision for some other moderator.
 202         token, token_owner, member = self._registrar.register(self._anne)
 203         # Anne's subscription request got held.
 204         self.assertIsNone(member)
 205         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 206         with self.assertRaises(HTTPError) as cm:
 207             call_api(url.format(token), dict(
 208                 action='defer',
 209                 ))
 210         self.assertEqual(cm.exception.code, 204)
 211         # Anne is not a member.
 212         self.assertIsNone(self._mlist.get_members('anne@example.com'))
 213         # The request URL still exists.
 214         with self.assertRaises(HTTPError) as cm:
 215             call_api(url.format(token), dict(
 216                 action='defer',
 217                 ))
 218         self.assertEqual(cm.exception.code, 204)
 219         # And now we can accept it.
 220         with self.assertRaises(HTTPError) as cm:
 221             call_api(url.format(token), dict(
 222                 action='accept',
 223                 ))
 224         self.assertEqual(cm.exception.code, 204)
 225         # Anne is a member.
 226         self.assertEqual(
 227             self._mlist.get_members('anne@example.com').address,
 228             self._anne)
 229         # The request URL no longer exists.
 230         with self.assertRaises(HTTPError) as cm:
 231             call_api(url.format(token), dict(
 232                 action='accept',
 233                 ))
 234         self.assertEqual(cm.exception.code, 404)
 235
 236     def test_reject(self):
 237         # POST to the request to reject it.  This leaves a bounce message in
 238         # the virgin queue.
 239         token, token_owner, member = self._registrar.register(self._anne)
 240         # Anne's subscription request got held.
 241         self.assertIsNone(member)
 242         # There are currently no messages in the virgin queue.
 243         items = get_queue_messages('virgin')
 244         self.assertEqual(len(items), 0)
 245         url = 'http://localhost:9001/3.0/lists/ant@example.com/requests/{}'
 246         with self.assertRaises(HTTPError) as cm:
 247             call_api(url.format(token), dict(
 248                 action='reject',
 249                 ))
 250         self.assertEqual(cm.exception.code, 204)
 251         # Anne is not a member.
 252         self.assertIsNone(self._mlist.get_members('anne@example.com'))
 253         # The request URL no longer exists.
 254         with self.assertRaises(HTTPError) as cm:
 255             call_api(url.format(token), dict(
 256                 action='reject',
 257                 ))
 258         self.assertEqual(cm.exception.code, 404)
 259         # And the rejection message is now in the virgin queue.
 260         items = get_queue_messages('virgin')
 261         self.assertEqual(len(items), 1)
 262         self.assertEqual(str(items[0].msg), '')