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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for branch merge queue collections."""
__metaclass__ = type
from zope.component import getUtility
from zope.security.proxy import removeSecurityProxy
from canonical.launchpad.interfaces.lpstorm import IMasterStore
from canonical.testing.layers import DatabaseFunctionalLayer
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.code.interfaces.branchmergequeuecollection import (
IAllBranchMergeQueues,
IBranchMergeQueueCollection,
)
from lp.code.interfaces.codehosting import LAUNCHPAD_SERVICES
from lp.code.model.branchmergequeue import BranchMergeQueue
from lp.code.model.branchmergequeuecollection import (
GenericBranchMergeQueueCollection,
)
from lp.testing import TestCaseWithFactory
class TestGenericBranchMergeQueueCollection(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
TestCaseWithFactory.setUp(self)
self.store = IMasterStore(BranchMergeQueue)
def test_provides_branchmergequeuecollection(self):
# `GenericBranchMergeQueueCollection`
# provides the `IBranchMergeQueueCollection` interface.
self.assertProvides(
GenericBranchMergeQueueCollection(self.store),
IBranchMergeQueueCollection)
def test_getMergeQueues_no_filter_no_queues(self):
# If no filter is specified, then the collection is of all branches
# merge queues. By default, there are no branch merge queues.
collection = GenericBranchMergeQueueCollection(self.store)
self.assertEqual([], list(collection.getMergeQueues()))
def test_getMergeQueues_no_filter(self):
# If no filter is specified, then the collection is of all branch
# merge queues.
collection = GenericBranchMergeQueueCollection(self.store)
queue = self.factory.makeBranchMergeQueue()
self.assertEqual([queue], list(collection.getMergeQueues()))
def test_count(self):
# The 'count' property of a collection is the number of elements in
# the collection.
collection = GenericBranchMergeQueueCollection(self.store)
self.assertEqual(0, collection.count())
for i in range(3):
self.factory.makeBranchMergeQueue()
self.assertEqual(3, collection.count())
def test_count_respects_filter(self):
# If a collection is a subset of all possible queues, then the count
# will be the size of that subset. That is, 'count' respects any
# filters that are applied.
person = self.factory.makePerson()
queue = self.factory.makeBranchMergeQueue(owner=person)
queue2 = self.factory.makeAnyBranch()
collection = GenericBranchMergeQueueCollection(
self.store, [BranchMergeQueue.owner == person])
self.assertEqual(1, collection.count())
class TestBranchMergeQueueCollectionFilters(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
TestCaseWithFactory.setUp(self)
self.all_queues = getUtility(IAllBranchMergeQueues)
def test_count_respects_visibleByUser_filter(self):
# IBranchMergeQueueCollection.count() returns the number of queues
# that getMergeQueues() yields, even when the visibleByUser filter is
# applied.
branch = self.factory.makeAnyBranch(private=True)
naked_branch = removeSecurityProxy(branch)
queue = self.factory.makeBranchMergeQueue(branches=[naked_branch])
branch2 = self.factory.makeAnyBranch(private=True)
naked_branch2 = removeSecurityProxy(branch2)
queue2 = self.factory.makeBranchMergeQueue(branches=[naked_branch2])
collection = self.all_queues.visibleByUser(naked_branch.owner)
self.assertEqual(1, len(collection.getMergeQueues()))
self.assertEqual(1, collection.count())
def test_ownedBy(self):
# 'ownedBy' returns a new collection restricted to queues owned by
# the given person.
queue = self.factory.makeBranchMergeQueue()
queue2 = self.factory.makeBranchMergeQueue()
collection = self.all_queues.ownedBy(queue.owner)
self.assertEqual([queue], collection.getMergeQueues())
class TestGenericBranchMergeQueueCollectionVisibleFilter(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
TestCaseWithFactory.setUp(self)
public_branch = self.factory.makeAnyBranch(name='public')
self.queue_with_public_branch = self.factory.makeBranchMergeQueue(
branches=[removeSecurityProxy(public_branch)])
private_branch1 = self.factory.makeAnyBranch(
private=True, name='private1')
naked_private_branch1 = removeSecurityProxy(private_branch1)
self.private_branch1_owner = naked_private_branch1.owner
self.queue1_with_private_branch = self.factory.makeBranchMergeQueue(
branches=[naked_private_branch1])
private_branch2 = self.factory.makeAnyBranch(
private=True, name='private2')
self.queue2_with_private_branch = self.factory.makeBranchMergeQueue(
branches=[removeSecurityProxy(private_branch2)])
self.all_queues = getUtility(IAllBranchMergeQueues)
def test_all_queues(self):
# Without the visibleByUser filter, all queues are in the
# collection.
self.assertEqual(
sorted([self.queue_with_public_branch,
self.queue1_with_private_branch,
self.queue2_with_private_branch]),
sorted(self.all_queues.getMergeQueues()))
def test_anonymous_sees_only_public(self):
# Anonymous users can see only queues with public branches.
queues = self.all_queues.visibleByUser(None)
self.assertEqual([self.queue_with_public_branch],
list(queues.getMergeQueues()))
def test_random_person_sees_only_public(self):
# Logged in users with no special permissions can see only queues with
# public branches.
person = self.factory.makePerson()
queues = self.all_queues.visibleByUser(person)
self.assertEqual([self.queue_with_public_branch],
list(queues.getMergeQueues()))
def test_owner_sees_own_branches(self):
# Users can always see the queues with branches that they own, as well
# as queues with public branches.
queues = self.all_queues.visibleByUser(self.private_branch1_owner)
self.assertEqual(
sorted([self.queue_with_public_branch,
self.queue1_with_private_branch]),
sorted(queues.getMergeQueues()))
def test_owner_member_sees_own_queues(self):
# Members of teams that own queues can see queues owned by those
# teams, as well as public branches.
team_owner = self.factory.makePerson()
team = self.factory.makeTeam(team_owner)
private_branch = self.factory.makeAnyBranch(
owner=team, private=True, name='team')
queue_with_private_branch = self.factory.makeBranchMergeQueue(
branches=[removeSecurityProxy(private_branch)])
queues = self.all_queues.visibleByUser(team_owner)
self.assertEqual(
sorted([self.queue_with_public_branch,
queue_with_private_branch]),
sorted(queues.getMergeQueues()))
def test_launchpad_services_sees_all(self):
# The LAUNCHPAD_SERVICES special user sees *everything*.
queues = self.all_queues.visibleByUser(LAUNCHPAD_SERVICES)
self.assertEqual(
sorted(self.all_queues.getMergeQueues()),
sorted(queues.getMergeQueues()))
def test_admins_see_all(self):
# Launchpad administrators see *everything*.
admin = self.factory.makePerson()
admin_team = removeSecurityProxy(
getUtility(ILaunchpadCelebrities).admin)
admin_team.addMember(admin, admin_team.teamowner)
queues = self.all_queues.visibleByUser(admin)
self.assertEqual(
sorted(self.all_queues.getMergeQueues()),
sorted(queues.getMergeQueues()))
|