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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
# Copyright 2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Connect feature flags into scopes where they can be used.
The most common is flags scoped by some attribute of a web request, such as
the page ID or the server name. But other types of scope can also match code
run from cron scripts and potentially also other places.
"""
__all__ = [
'HANDLERS',
'ScopesForScript',
'ScopesFromRequest',
'undocumented_scopes',
]
__metaclass__ = type
import re
from zope.component import getUtility
from canonical.launchpad.webapp.interfaces import ILaunchBag
from lp.services.propertycache import cachedproperty
import canonical.config
undocumented_scopes = set()
class BaseScope():
"""A base class for scope handlers.
The docstring of subclasses is used on the +feature-info page as
documentation, so write them accordingly.
"""
# The regex pattern used to decide if a handler can evaluate a particular
# scope. Also used on +feature-info.
pattern = None
@cachedproperty
def compiled_pattern(self):
"""The compiled scope matching regex. A small optimization."""
return re.compile(self.pattern)
def lookup(self, scope_name):
"""Returns true if the given scope name is "active"."""
raise NotImplementedError('Subclasses of BaseScope must implement '
'lookup.')
class DefaultScope(BaseScope):
"""The default scope. Always active."""
pattern = r'default$'
def lookup(self, scope_name):
return True
class BaseWebRequestScope(BaseScope):
"""Base class for scopes that key off web request attributes."""
def __init__(self, request):
self.request = request
class PageScope(BaseWebRequestScope):
"""The current page ID.
Pageid scopes are written as 'pageid:' + the pageid to match. Pageids
are treated as a namespace with : and # delimiters.
For example, the scope 'pageid:Foo' will be active on pages with pageids:
Foo
Foo:Bar
Foo#quux
"""
pattern = r'pageid:'
def lookup(self, scope_name):
"""Is the given scope match the current pageid?"""
pageid_scope = scope_name[len('pageid:'):]
scope_segments = self._pageid_to_namespace(pageid_scope)
request_segments = self._request_pageid_namespace
# In 2.6, this can be replaced with izip_longest
for pos, name in enumerate(scope_segments):
if pos == len(request_segments):
return False
if request_segments[pos] != name:
return False
return True
@staticmethod
def _pageid_to_namespace(pageid):
"""Return a list of namespace elements for pageid."""
# Normalise delimiters.
pageid = pageid.replace('#', ':')
# Create a list to walk, empty namespaces are elided.
return [name for name in pageid.split(':') if name]
@cachedproperty
def _request_pageid_namespace(self):
return tuple(self._pageid_to_namespace(
self.request._orig_env.get('launchpad.pageid', '')))
class TeamScope(BaseScope):
"""The current user's team memberships.
Team ID scopes are written as 'team:' + the team name to match.
The scope 'team:launchpad-beta-users' will match members of the team
'launchpad-beta-users'.
"""
pattern = r'team:'
def lookup(self, scope_name):
"""Is the given scope a team membership?
This will do a two queries, so we probably want to keep the number of
team based scopes in use to a small number. (Person.inTeam could be
fixed to reduce this to one query).
"""
team_name = scope_name[len('team:'):]
person = getUtility(ILaunchBag).user
if person is None:
return False
return person.inTeam(team_name)
class ServerScope(BaseScope):
"""Matches the current server.
For example, the scope server.lpnet is active when is_lpnet is set to True
in the Launchpad configuration.
"""
pattern = r'server\.'
def lookup(self, scope_name):
"""Match the current server as a scope."""
server_name = scope_name.split('.', 1)[1]
try:
return canonical.config.config['launchpad']['is_' + server_name]
except KeyError:
pass
return False
class ScriptScope(BaseScope):
"""Matches the name of the currently running script.
For example, the scope script:embroider is active in a script called
"embroider."
"""
pattern = r'script:'
def __init__(self, script_name):
self.script_scope = self.pattern + script_name
def lookup(self, scope_name):
"""Match the running script as a scope."""
return scope_name == self.script_scope
# These are the handlers for all of the allowable scopes, listed here so that
# we can for example show all of them in an admin page. Any new scope will
# need a scope handler and that scope handler has to be added to this list.
# See BaseScope for hints as to what a scope handler should look like.
HANDLERS = set([DefaultScope, PageScope, TeamScope, ServerScope, ScriptScope])
class MultiScopeHandler():
"""A scope handler that combines multiple `BaseScope`s.
The ordering in which they're added is arbitrary, because precedence is
determined by the ordering of rules.
"""
def __init__(self, scopes):
self.handlers = scopes
def _findMatchingHandlers(self, scope_name):
"""Find any handlers that match `scope_name`."""
return [
handler
for handler in self.handlers
if handler.compiled_pattern.match(scope_name)]
def lookup(self, scope_name):
"""Determine if scope_name applies.
This method iterates over the configured scope handlers until it
either finds one that claims the requested scope name matches,
or the handlers are exhausted, in which case the
scope name is not a match.
"""
matching_handlers = self._findMatchingHandlers(scope_name)
for handler in matching_handlers:
if handler.lookup(scope_name):
return True
# If we didn't find at least one matching handler, then the
# requested scope is unknown and we want to record the scope for
# the +flag-info page to display.
if len(matching_handlers) == 0:
undocumented_scopes.add(scope_name)
class ScopesFromRequest(MultiScopeHandler):
"""Identify feature scopes based on request state."""
def __init__(self, request):
super(ScopesFromRequest, self).__init__([
DefaultScope(),
PageScope(request),
TeamScope(),
ServerScope()])
class ScopesForScript(MultiScopeHandler):
"""Identify feature scopes for a given script."""
def __init__(self, script_name):
super(ScopesForScript, self).__init__([
DefaultScope(),
ScriptScope(script_name)])
|