Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Newer
Older
100644 254 lines (220 sloc) 10.221 kb
8e7fcc5 @gnarea A huge commit to fix three thread-unsafe predicates
gnarea authored
1 # -*- coding: utf-8 -*-
2 ##############################################################################
3 #
4 # Copyright (c) 2009, Gustavo Narea <me@gustavonarea.net>
5 # All Rights Reserved.
6 #
7 # This software is subject to the provisions of the BSD-like license at
8 # http://www.repoze.org/LICENSE.txt. A copy of the license should accompany
9 # this distribution. THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL
10 # EXPRESS OR IMPLIED WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO,
11 # THE IMPLIED WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND
12 # FITNESS FOR A PARTICULAR PURPOSE.
13 #
14 ##############################################################################
15
16 """
17 Ensure :mod:`repoze.what` is thread-safe when some components are instantiated
18 at module-level and shared among threads.
19
20 Thanks to Alberto Valverde for helping me write these tests!
21
22 """
23
24 import sys, time
25 from threading import Thread, Event, Timer
26 from string import ascii_letters
27 from unittest import TestCase
28
29 from repoze.what.predicates import *
30 from repoze.what.authorize import check_authorization, NotAuthorizedError
31
32
33 #{ The test cases
34
35
36 class TestPredicateErrors(TestCase):
37 """
38 Test that all the built-in predicates are thread-safe.
39
40 Alberto Valverde found that "All"-based predicates were not thread-safe
41 because if they are instantiated at module-level and used among many
42 threads, their error message of a predicate that failed on one thread would
43 override the message of a failed predicate on another thread -- that's
44 because there'd be only one predicate, not one per thread.
45
46 So this test case would ensure this won't happen again, on any other
47 predicate, "All"-based or not.
48
49 """
50
51 def setUp(self):
52 self.found_error = Event()
53 self.threads = []
54 self.threads_stopped = False
55
56 def tearDown(self):
57 self._stop_threads()
58
59 def _share_predicate_among_threads(self, shared_predicate, scenarios):
60 """
61 Share predicate ``shared_predicate`` among many threads (one per
62 scenario).
63
64 """
65 # First define the scenarios to be run in one thread
66 for scenario in scenarios:
67 thread = DaPredicateThread(scenario['credentials'],
68 shared_predicate, scenario['error'],
69 self.found_error)
70 self.threads.append(thread)
71 # Configuring the timer that is going to stop threads when the
72 # predicate in question IS thread-safe.
73 t = Timer(2, self._stop_threads)
74 t.start()
75 # Running the threads:
76 map(Thread.start, self.threads)
77 if self.found_error.isSet():
78 predicate_class = shared_predicate.__class__.__name__
79 self.fail('Predicate %s is not thread-safe' % predicate_class)
80
81 def _stop_threads(self):
82 """Stop all the threads"""
83 if self.threads_stopped:
84 return
85 for thread in self.threads:
86 thread.stop = True
87 self.threads_stopped = True
88
89 def test_is_user(self):
90 """The is_user predicate is thread-safe."""
91 # Building the shared predicate:
92 users = set(ascii_letters)
93 shared_predicate = is_user('foo')
94 error = 'The current user must be "foo"'
95 # Building the test scenarios that will share the predicate above:
96 scenarios = []
97 for u in users:
98 credentials = {'repoze.what.userid': u}
99 scenario = {'credentials': credentials, 'error': error}
100 scenarios.append(scenario)
101 self._share_predicate_among_threads(shared_predicate, scenarios)
102
103 def test_in_group(self):
104 """The in_group predicate is thread-safe."""
105 # Building the shared predicate:
106 groups = set(ascii_letters)
107 shared_predicate = in_group('foo')
108 error = 'The current user must belong to the group "foo"'
109 # Building the test scenarios that will share the predicate above:
110 scenarios = []
111 for g in groups:
112 credentials = {'groups': groups.copy()}
113 scenario = {'credentials': credentials, 'error': error}
114 scenarios.append(scenario)
115 self._share_predicate_among_threads(shared_predicate, scenarios)
116
117 def test_in_all_groups(self):
118 """The in_all_groups predicate is thread-safe."""
119 # Building the shared predicate:
120 all_groups = set(ascii_letters)
121 shared_predicate = in_all_groups(*all_groups)
122 # Building the test scenarios that will share the predicate above:
123 scenarios = []
124 for g in all_groups:
125 error = 'The current user must belong to the group "%s"' % g
126 scenario = {
127 'credentials': {'groups': all_groups - set([g])},
128 'error': error,
129 }
130 scenarios.append(scenario)
131 self._share_predicate_among_threads(shared_predicate, scenarios)
132
133 def test_in_any_group(self):
134 """The in_any_group predicate is thread-safe."""
135 # Building the shared predicate:
136 all_groups = set(ascii_letters)
137 shared_predicate = in_any_group(*all_groups)
138 error = "The member must belong to at least one of the following "\
139 "groups: "
140 error = error + ', '.join(all_groups)
141 # Building the test scenarios that will share the predicate above:
142 credentials = {'groups': set([u"ñ", u"é"])}
143 scenarios = []
144 for g in all_groups:
145 scenario = {'credentials': credentials, 'error': error}
146 scenarios.append(scenario)
147 self._share_predicate_among_threads(shared_predicate, scenarios)
148
149 def test_has_permission(self):
150 """The has_permission predicate is thread-safe."""
151 # Building the shared predicate:
152 perms = set(ascii_letters)
153 shared_predicate = has_permission('foo')
154 error = 'The user must have the "foo" permission'
155 # Building the test scenarios that will share the predicate above:
156 scenarios = []
157 for p in perms:
158 credentials = {'permissions': perms.copy()}
159 scenario = {'credentials': credentials, 'error': error}
160 scenarios.append(scenario)
161 self._share_predicate_among_threads(shared_predicate, scenarios)
162
163 def test_has_all_permissions(self):
164 """The has_all_permissions predicate is thread-safe."""
165 # Building the shared predicate:
166 all_perms = set(ascii_letters)
167 shared_predicate = has_all_permissions(*all_perms)
168 # Building the test scenarios that will share the predicate above:
169 scenarios = []
170 for p in all_perms:
171 error = 'The user must have the "%s" permission' % p
172 scenario = {
173 'credentials': {'permissions': all_perms - set([p])},
174 'error': error,
175 }
176 scenarios.append(scenario)
177 self._share_predicate_among_threads(shared_predicate, scenarios)
178
179 def test_has_any_permission(self):
180 """The has_any_permission predicate is thread-safe."""
181 # Building the shared predicate:
182 all_perms = set(ascii_letters)
183 shared_predicate = has_any_permission(*all_perms)
184 error = "The user must have at least one of the following "\
185 "permissions: "
186 error = error + ', '.join(all_perms)
187 # Building the test scenarios that will share the predicate above:
188 credentials = {'permissions': set([u"ñ", u"é"])}
189 scenarios = []
190 for p in all_perms:
191 scenario = {'credentials': credentials, 'error': error}
192 scenarios.append(scenario)
193 self._share_predicate_among_threads(shared_predicate, scenarios)
194
195 def test_All(self):
196 """The All predicate is thread-safe."""
197 # Building the shared predicate:
198 all_groups = set(ascii_letters)
199 shared_predicate = All(*map(in_group, all_groups))
200 # Building the test scenarios that will share the predicate above:
201 scenarios = []
202 for g in all_groups:
203 error = 'The current user must belong to the group "%s"' % g
204 scenario = {
205 'credentials': {'groups': all_groups - set([g])},
206 'error': error,
207 }
208 scenarios.append(scenario)
209 self._share_predicate_among_threads(shared_predicate, scenarios)
210
211 def test_Any(self):
212 """The Any predicate is thread-safe."""
213 # Building the shared predicate:
214 expected_users = set(ascii_letters)
215 shared_predicate = Any(*map(is_user, expected_users))
216 error = "At least one of the following predicates must be met:"
217 for u in expected_users:
218 error = error + ' The current user must be "%s",' % u
219 error = error[:-1]
220 # Building the test scenarios that will share the predicate above:
221 credentials = {'repoze.what.userid': None}
222 scenarios = []
223 for u in expected_users:
224 scenario = {'credentials': credentials, 'error': error}
225 scenarios.append(scenario)
226 self._share_predicate_among_threads(shared_predicate, scenarios)
227
228
229 #{ Test utilities
230
231
232 class DaPredicateThread(Thread):
233 def __init__(self, credentials, shared_predicate, expected_error,
234 found_error, *args, **kwargs):
235 super(DaPredicateThread, self).__init__(*args, **kwargs)
236 self.credentials = credentials
237 self.shared_predicate = shared_predicate
238 self.expected_error = expected_error
239 self.found_error = found_error
240 self.stop = False
241
242 def run(self):
243 while not self.found_error.isSet() and not self.stop:
244 # Create a new environ simulating the fresh environ each request gets
245 environ = {'repoze.what.credentials': self.credentials.copy()}
246 try:
247 check_authorization(self.shared_predicate, environ)
248 except NotAuthorizedError, exc:
249 if unicode(exc) != self.expected_error:
250 self.found_error.set()
251
252
253 #}
Something went wrong with that request. Please try again.