167
167
"""Tests for acquiring branches to pull.
169
169
The tests apply to branches accessed directly or through an XML-RPC style
170
endpoint -- implement `assertNoBranchIsAquired`, `assertBranchIsAquired`
170
endpoint -- implement `assertNoBranchIsAcquired`, `assertBranchIsAcquired`
171
171
and `startMirroring` as appropriate.
174
def assertNoBranchIsAquired(self, *branch_types):
174
def assertNoBranchIsAcquired(self, *branch_types):
175
175
"""Assert that there is no branch to pull.
177
177
:param branch_types: A list of branch types to pass to
178
178
acquireBranchToPull. Passing none means consider all types of
181
raise NotImplementedError(self.assertNoBranchIsAquired)
181
raise NotImplementedError(self.assertNoBranchIsAcquired)
183
def assertBranchIsAquired(self, branch, *branch_types):
183
def assertBranchIsAcquired(self, branch, *branch_types):
184
184
"""Assert that ``branch`` is the next branch to be pulled.
186
186
:param branch_types: A list of branch types to pass to
187
187
acquireBranchToPull. Passing none means consider all types of
190
raise NotImplementedError(self.assertBranchIsAquired)
190
raise NotImplementedError(self.assertBranchIsAcquired)
192
192
def startMirroring(self, branch):
193
193
"""Mark that ``branch`` has begun mirroring."""
196
196
def test_empty(self):
197
197
# If there is no branch that needs pulling, acquireBranchToPull
199
self.assertNoBranchIsAquired()
199
self.assertNoBranchIsAcquired()
201
201
def test_simple(self):
202
202
# If there is one branch that needs mirroring, acquireBranchToPull
204
204
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
205
205
branch.requestMirror()
206
self.assertBranchIsAquired(branch)
206
self.assertBranchIsAcquired(branch)
208
208
def test_remote_branch_not_acquired(self):
209
209
# On a few occasions a branch type that is mirrored has been
212
212
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
213
213
branch.requestMirror()
214
214
removeSecurityProxy(branch).branch_type = BranchType.REMOTE
215
self.assertNoBranchIsAquired()
215
self.assertNoBranchIsAcquired()
217
217
def test_private(self):
218
218
# If there is a private branch that needs mirroring,
220
220
branch = self.factory.makeAnyBranch(
221
221
branch_type=BranchType.MIRRORED, private=True)
222
222
removeSecurityProxy(branch).requestMirror()
223
self.assertBranchIsAquired(branch)
223
self.assertBranchIsAcquired(branch)
225
225
def test_no_inprogress(self):
226
226
# If a branch is being mirrored, it is not returned.
227
227
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
228
228
branch.requestMirror()
229
229
self.startMirroring(branch)
230
self.assertNoBranchIsAquired()
230
self.assertNoBranchIsAcquired()
232
232
def test_first_requested_returned(self):
233
233
# If two branches are to be mirrored, the one that was requested first
244
244
second_branch.requestMirror()
245
245
naked_second_branch = removeSecurityProxy(second_branch)
246
246
naked_second_branch.next_mirror_time -= timedelta(seconds=50)
247
self.assertBranchIsAquired(naked_first_branch)
247
self.assertBranchIsAcquired(naked_first_branch)
249
249
def test_type_filter_mirrrored_returns_mirrored(self):
250
250
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
251
251
branch.requestMirror()
252
self.assertBranchIsAquired(branch, BranchType.MIRRORED)
252
self.assertBranchIsAcquired(branch, BranchType.MIRRORED)
254
254
def test_type_filter_imported_does_not_return_mirrored(self):
255
255
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
256
256
branch.requestMirror()
257
self.assertNoBranchIsAquired(BranchType.IMPORTED)
257
self.assertNoBranchIsAcquired(BranchType.IMPORTED)
259
259
def test_type_filter_mirrored_imported_returns_mirrored(self):
260
260
branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
261
261
branch.requestMirror()
262
self.assertBranchIsAquired(
262
self.assertBranchIsAcquired(
263
263
branch, BranchType.MIRRORED, BranchType.IMPORTED)
265
265
def test_type_filter_mirrored_imported_returns_imported(self):
266
266
branch = self.factory.makeAnyBranch(branch_type=BranchType.IMPORTED)
267
267
branch.requestMirror()
268
self.assertBranchIsAquired(
268
self.assertBranchIsAcquired(
269
269
branch, BranchType.MIRRORED, BranchType.IMPORTED)
276
276
layer = DatabaseFunctionalLayer
278
def assertNoBranchIsAquired(self, *branch_types):
278
def assertNoBranchIsAcquired(self, *branch_types):
279
279
"""See `AcquireBranchToPullTests`."""
280
280
acquired_branch = getUtility(IBranchPuller).acquireBranchToPull(
282
282
self.assertEqual(None, acquired_branch)
284
def assertBranchIsAquired(self, branch, *branch_types):
284
def assertBranchIsAcquired(self, branch, *branch_types):
285
285
"""See `AcquireBranchToPullTests`."""
286
286
acquired_branch = getUtility(IBranchPuller).acquireBranchToPull(