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
|
# Copyright 2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Unit tests for methods of BranchMergeQueue."""
from __future__ import with_statement
import simplejson
from canonical.launchpad.interfaces.lpstorm import IStore
from canonical.launchpad.webapp.testing import verifyObject
from canonical.testing.layers import (
AppServerLayer,
DatabaseFunctionalLayer,
)
from lp.code.errors import InvalidMergeQueueConfig
from lp.code.interfaces.branchmergequeue import IBranchMergeQueue
from lp.code.model.branchmergequeue import BranchMergeQueue
from lp.testing import (
ANONYMOUS,
person_logged_in,
launchpadlib_for,
TestCaseWithFactory,
ws_object,
)
class TestBranchMergeQueueInterface(TestCaseWithFactory):
"""Test IBranchMergeQueue interface."""
layer = DatabaseFunctionalLayer
def test_implements_interface(self):
queue = self.factory.makeBranchMergeQueue()
IStore(BranchMergeQueue).add(queue)
verifyObject(IBranchMergeQueue, queue)
class TestBranchMergeQueueSource(TestCaseWithFactory):
"""Test the methods of IBranchMergeQueueSource."""
layer = DatabaseFunctionalLayer
def test_new(self):
owner = self.factory.makePerson()
name = u'SooperQueue'
description = u'This is Sooper Queue'
config = unicode(simplejson.dumps({'test': 'make check'}))
queue = BranchMergeQueue.new(
name, owner, owner, description, config)
self.assertEqual(queue.name, name)
self.assertEqual(queue.owner, owner)
self.assertEqual(queue.registrant, owner)
self.assertEqual(queue.description, description)
self.assertEqual(queue.configuration, config)
class TestBranchMergeQueue(TestCaseWithFactory):
"""Test the functions of the BranchMergeQueue."""
layer = DatabaseFunctionalLayer
def test_branches(self):
"""Test that a merge queue can get all its managed branches."""
store = IStore(BranchMergeQueue)
queue = self.factory.makeBranchMergeQueue()
store.add(queue)
branch = self.factory.makeBranch()
store.add(branch)
with person_logged_in(branch.owner):
branch.addToQueue(queue)
self.assertEqual(
list(queue.branches),
[branch])
def test_setMergeQueueConfig(self):
"""Test that the configuration is set properly."""
queue = self.factory.makeBranchMergeQueue()
config = unicode(simplejson.dumps({
'test': 'make test'}))
with person_logged_in(queue.owner):
queue.setMergeQueueConfig(config)
self.assertEqual(queue.configuration, config)
def test_setMergeQueueConfig_invalid_json(self):
"""Test that invalid json can't be set as the config."""
queue = self.factory.makeBranchMergeQueue()
with person_logged_in(queue.owner):
self.assertRaises(
InvalidMergeQueueConfig,
queue.setMergeQueueConfig,
'abc')
class TestWebservice(TestCaseWithFactory):
layer = AppServerLayer
def test_properties(self):
"""Test that the correct properties are exposed."""
with person_logged_in(ANONYMOUS):
name = u'teh-queue'
description = u'Oh hai! I are a queues'
configuration = unicode(simplejson.dumps({'test': 'make check'}))
queuer = self.factory.makePerson()
db_queue = self.factory.makeBranchMergeQueue(
registrant=queuer, owner=queuer, name=name,
description=description,
configuration=configuration)
branch1 = self.factory.makeBranch()
with person_logged_in(branch1.owner):
branch1.addToQueue(db_queue)
branch2 = self.factory.makeBranch()
with person_logged_in(branch2.owner):
branch2.addToQueue(db_queue)
launchpad = launchpadlib_for('test', db_queue.owner,
service_root="http://api.launchpad.dev:8085")
queuer = ws_object(launchpad, queuer)
queue = ws_object(launchpad, db_queue)
branch1 = ws_object(launchpad, branch1)
branch2 = ws_object(launchpad, branch2)
self.assertEqual(queue.registrant, queuer)
self.assertEqual(queue.owner, queuer)
self.assertEqual(queue.name, name)
self.assertEqual(queue.description, description)
self.assertEqual(queue.configuration, configuration)
self.assertEqual(queue.date_created, db_queue.date_created)
self.assertEqual(len(queue.branches), 2)
def test_set_configuration(self):
"""Test the mutator for setting configuration."""
with person_logged_in(ANONYMOUS):
db_queue = self.factory.makeBranchMergeQueue()
launchpad = launchpadlib_for('test', db_queue.owner,
service_root="http://api.launchpad.dev:8085")
configuration = simplejson.dumps({'test': 'make check'})
queue = ws_object(launchpad, db_queue)
queue.configuration = configuration
queue.lp_save()
queue2 = ws_object(launchpad, db_queue)
self.assertEqual(queue2.configuration, configuration)
|