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
|
# Copyright 2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Implementations of `IBranchMergeQueueCollection`."""
__metaclass__ = type
__all__ = [
'GenericBranchCollection',
]
from zope.interface import implements
from canonical.launchpad.interfaces.lpstorm import IMasterStore
from lp.code.interfaces.branchmergequeue import (
user_has_special_merge_queue_access,
)
from lp.code.interfaces.branchmergequeuecollection import (
IBranchMergeQueueCollection,
InvalidFilter,
)
from lp.code.interfaces.codehosting import LAUNCHPAD_SERVICES
from lp.code.model.branchmergequeue import BranchMergeQueue
class GenericBranchMergeQueueCollection:
"""See `IBranchMergeQueueCollection`."""
implements(IBranchMergeQueueCollection)
def __init__(self, store=None, merge_queue_filter_expressions=None,
tables=None, exclude_from_search=None):
"""Construct a `GenericBranchMergeQueueCollection`.
:param store: The store to look in for merge queues. If not specified,
use the default store.
:param merge_queue_filter_expressions: A list of Storm expressions to
restrict the queues in the collection. If unspecified, then
there will be no restrictions on the result set. That is, all
queues in the store will be in the collection.
:param tables: A dict of Storm tables to the Join expression. If an
expression in merge_queue_filter_expressions refers to a table,
then that table *must* be in this list.
"""
self._store = store
if merge_queue_filter_expressions is None:
merge_queue_filter_expressions = []
self._merge_queue_filter_expressions = merge_queue_filter_expressions
if tables is None:
tables = {}
self._tables = tables
if exclude_from_search is None:
exclude_from_search = []
self._exclude_from_search = exclude_from_search
def count(self):
return self._getCount()
def _getCount(self):
"""See `IBranchMergeQueueCollection`."""
return self._getMergeQueues().count()
@property
def store(self):
if self._store is None:
return IMasterStore(BranchMergeQueue)
else:
return self._store
def _filterBy(self, expressions, table=None, join=None,
exclude_from_search=None):
"""Return a subset of this collection, filtered by 'expressions'."""
tables = self._tables.copy()
if table is not None:
if join is None:
raise InvalidFilter("Cannot specify a table without a join.")
tables[table] = join
if exclude_from_search is None:
exclude_from_search = []
if expressions is None:
expressions = []
return self.__class__(
self.store,
self._merge_queue_filter_expressions + expressions,
tables,
self._exclude_from_search + exclude_from_search)
def _getMergeQueueExpressions(self):
"""Return the where expressions for this collection."""
return self._merge_queue_filter_expressions
def getMergeQueues(self):
return list(self._getMergeQueues())
def _getMergeQueues(self):
"""See `IBranchMergeQueueCollection`."""
tables = [BranchMergeQueue] + self._tables.values()
expressions = self._getMergeQueueExpressions()
return self.store.using(*tables).find(BranchMergeQueue, *expressions)
def ownedBy(self, person):
"""See `IBranchMergeQueueCollection`."""
return self._filterBy([BranchMergeQueue.owner == person])
def visibleByUser(self, person):
"""See `IBranchMergeQueueCollection`."""
if (person == LAUNCHPAD_SERVICES or
user_has_special_merge_queue_access(person)):
return self
return VisibleBranchMergeQueueCollection(
person,
self._store, None,
self._tables, self._exclude_from_search)
class VisibleBranchMergeQueueCollection(GenericBranchMergeQueueCollection):
"""A mergequeue collection which provides queues visible by a user."""
def __init__(self, person, store=None,
merge_queue_filter_expressions=None, tables=None,
exclude_from_search=None):
super(VisibleBranchMergeQueueCollection, self).__init__(
store=store,
merge_queue_filter_expressions=merge_queue_filter_expressions,
tables=tables,
exclude_from_search=exclude_from_search,
)
self._user = person
def _filterBy(self, expressions, table=None, join=None,
exclude_from_search=None):
"""Return a subset of this collection, filtered by 'expressions'."""
tables = self._tables.copy()
if table is not None:
if join is None:
raise InvalidFilter("Cannot specify a table without a join.")
tables[table] = join
if exclude_from_search is None:
exclude_from_search = []
if expressions is None:
expressions = []
return self.__class__(
self._user,
self.store,
self._merge_queue_filter_expressions + expressions,
tables,
self._exclude_from_search + exclude_from_search)
def visibleByUser(self, person):
"""See `IBranchMergeQueueCollection`."""
if person == self._user:
return self
raise InvalidFilter(
"Cannot filter for merge queues visible by user %r, already "
"filtering for %r" % (person, self._user))
def _getCount(self):
"""See `IBranchMergeQueueCollection`."""
return len(self._getMergeQueues())
def _getMergeQueues(self):
"""Return the queues visible by self._user.
A queue is visible to a user if that user can see all the branches
associated with the queue.
"""
def allBranchesVisible(user, branches):
return len([branch for branch in branches
if branch.visibleByUser(user)]) == branches.count()
queues = super(
VisibleBranchMergeQueueCollection, self)._getMergeQueues()
return [queue for queue in queues
if allBranchesVisible(self._user, queue.branches)]
|