1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
# Copyright (C) 2012-2017 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""Test the various pending requests interfaces."""
import unittest
from contextlib import contextmanager
from itertools import count
from mailman.app.lifecycle import create_list
from mailman.app.moderator import hold_message
from mailman.config import config
from mailman.interfaces.requests import IListRequests, RequestType
from mailman.model.requests import _Request
from mailman.testing.helpers import specialized_message_from_string as mfs
from mailman.testing.layers import ConfigLayer
from sqlalchemy.event import listen, remove
@contextmanager
def before_flush(id_hacker):
listen(config.db.store, 'before_flush', id_hacker)
try:
yield
finally:
remove(config.db.store, 'before_flush', id_hacker)
class TestRequests(unittest.TestCase):
layer = ConfigLayer
def setUp(self):
self._mlist = create_list('ant@example.com')
self._requests_db = IListRequests(self._mlist)
self._msg = mfs("""\
From: anne@example.com
To: ant@example.com
Subject: Something
Message-ID: <alpha>
Something else.
""")
def test_get_request_with_type(self):
# get_request() takes an optional request type.
request_id = hold_message(self._mlist, self._msg)
# Submit a request with a non-matching type. This should return None
# as if there were no matches.
response = self._requests_db.get_request(
request_id, RequestType.subscription)
self.assertEqual(response, None)
# Submit the same request with a matching type.
key, data = self._requests_db.get_request(
request_id, RequestType.held_message)
self.assertEqual(key, '<alpha>')
# It should also succeed with no optional request type given.
key, data = self._requests_db.get_request(request_id)
self.assertEqual(key, '<alpha>')
def test_hold_with_bogus_type(self):
# Calling hold_request() with a bogus request type is an error.
with self.assertRaises(TypeError) as cm:
self._requests_db.hold_request(5, 'foo')
self.assertEqual(cm.exception.args[0], 5)
def test_delete_missing_request(self):
# Trying to delete a missing request is an error.
with self.assertRaises(KeyError) as cm:
self._requests_db.delete_request(801)
self.assertEqual(cm.exception.args[0], 801)
def test_only_return_this_lists_requests(self):
# Issue #161: get_requests() returns requests that are not specific to
# the mailing list in question.
request_id = hold_message(self._mlist, self._msg)
bee = create_list('bee@example.com')
self.assertIsNone(IListRequests(bee).get_request(request_id))
def test_request_order(self):
# Requests must be sorted in creation order.
#
# This test only "works" for PostgreSQL, in the sense that if you
# remove the fix in ../requests.py, it will still pass in SQLite.
# Apparently SQLite auto-sorts results by ID but PostgreSQL autosorts
# by insertion time. It's still worth keeping the test to prevent
# regressions.
#
# We modify the auto-incremented ids by listening to SQLAlchemy's
# flush event, and hacking all the _Request object id's to the next
# value in a descending counter.
request_ids = []
counter = count(200, -1)
def id_hacker(session, flush_context, instances): # noqa: E306
for instance in session.new:
if isinstance(instance, _Request):
instance.id = next(counter)
with before_flush(id_hacker):
for index in range(10):
msg = mfs(self._msg.as_string())
msg.replace_header('Message-ID', '<alpha{}>'.format(index))
request_ids.append(hold_message(self._mlist, msg))
config.db.store.flush()
# Make sure that our ID are not already sorted.
self.assertNotEqual(request_ids, sorted(request_ids))
# Get requests and check their order.
requests = self._requests_db.of_type(RequestType.held_message)
self.assertEqual([r.id for r in requests], sorted(request_ids))
|
