← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/branch_implementations/test_locking.py

merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/branch_implementations/test_locking.py
227
227
                          ('rc', 'ul', True),
228
228
                         ], self.locks)
229
229
 
 
230
    def test_lock_write_returns_None_refuses_token(self):
 
231
        branch = self.make_branch('b')
 
232
        token = branch.lock_write()
 
233
        try:
 
234
            if token is not None:
 
235
                # This test does not apply, because this lockable supports
 
236
                # tokens.
 
237
                return
 
238
            self.assertRaises(errors.TokenLockingNotSupported,
 
239
                              branch.lock_write, token='token')
 
240
        finally:
 
241
            branch.unlock()
 
242
 
 
243
    def test_reentering_lock_write_raises_on_token_mismatch(self):
 
244
        branch = self.make_branch('b')
 
245
        token = branch.lock_write()
 
246
        try:
 
247
            if token is None:
 
248
                # This test does not apply, because this lockable refuses
 
249
                # tokens.
 
250
                return
 
251
            different_branch_token = token + 'xxx'
 
252
            # Re-using the same lockable instance with a different branch token
 
253
            # will raise TokenMismatch.
 
254
            self.assertRaises(errors.TokenMismatch,
 
255
                              branch.lock_write,
 
256
                              token=different_branch_token)
 
257
        finally:
 
258
            branch.unlock()
 
259
 
 
260
    def test_lock_write_with_nonmatching_token(self):
 
261
        branch = self.make_branch('b')
 
262
        token = branch.lock_write()
 
263
        try:
 
264
            if token is None:
 
265
                # This test does not apply, because this branch refuses
 
266
                # tokens.
 
267
                return
 
268
            different_branch_token = token + 'xxx'
 
269
 
 
270
            new_branch = branch.bzrdir.open_branch()
 
271
            # We only want to test the relocking abilities of branch, so use the
 
272
            # existing repository object which is already locked.
 
273
            new_branch.repository = branch.repository
 
274
            self.assertRaises(errors.TokenMismatch,
 
275
                              new_branch.lock_write,
 
276
                              token=different_branch_token)
 
277
        finally:
 
278
            branch.unlock()
 
279
 
 
280
 
 
281
    def test_lock_write_with_matching_token(self):
 
282
        """Test that a branch can be locked with a token, if it is already
 
283
        locked by that token."""
 
284
        branch = self.make_branch('b')
 
285
        token = branch.lock_write()
 
286
        try:
 
287
            if token is None:
 
288
                # This test does not apply, because this branch refuses tokens.
 
289
                return
 
290
            # The same instance will accept a second lock_write if the specified
 
291
            # token matches.
 
292
            branch.lock_write(token=token)
 
293
            branch.unlock()
 
294
            # Calling lock_write on a new instance for the same lockable will
 
295
            # also succeed.
 
296
            new_branch = branch.bzrdir.open_branch()
 
297
            # We only want to test the relocking abilities of branch, so use the
 
298
            # existing repository object which is already locked.
 
299
            new_branch.repository = branch.repository
 
300
            new_branch.lock_write(token=token)
 
301
            new_branch.unlock()
 
302
        finally:
 
303
            branch.unlock()
 
304
 
 
305
    def test_unlock_after_lock_write_with_token(self):
 
306
        # If lock_write did not physically acquire the lock (because it was
 
307
        # passed some tokens), then unlock should not physically release it.
 
308
        branch = self.make_branch('b')
 
309
        token = branch.lock_write()
 
310
        try:
 
311
            if token is None:
 
312
                # This test does not apply, because this lockable refuses
 
313
                # tokens.
 
314
                return
 
315
            new_branch = branch.bzrdir.open_branch()
 
316
            # We only want to test the relocking abilities of branch, so use the
 
317
            # existing repository object which is already locked.
 
318
            new_branch.repository = branch.repository
 
319
            new_branch.lock_write(token=token)
 
320
            new_branch.unlock()
 
321
            self.assertTrue(branch.get_physical_lock_status()) #XXX
 
322
        finally:
 
323
            branch.unlock()
 
324
 
 
325
    def test_lock_write_with_token_fails_when_unlocked(self):
 
326
        # First, lock and then unlock to get superficially valid tokens.  This
 
327
        # mimics a likely programming error, where a caller accidentally tries
 
328
        # to lock with a token that is no longer valid (because the original
 
329
        # lock was released).
 
330
        branch = self.make_branch('b')
 
331
        token = branch.lock_write()
 
332
        branch.unlock()
 
333
        if token is None:
 
334
            # This test does not apply, because this lockable refuses
 
335
            # tokens.
 
336
            return
 
337
 
 
338
        self.assertRaises(errors.TokenMismatch,
 
339
                          branch.lock_write, token=token)
 
340
 
 
341
    def test_lock_write_reenter_with_token(self):
 
342
        branch = self.make_branch('b')
 
343
        token = branch.lock_write()
 
344
        try:
 
345
            if token is None:
 
346
                # This test does not apply, because this lockable refuses
 
347
                # tokens.
 
348
                return
 
349
            # Relock with a token.
 
350
            branch.lock_write(token=token)
 
351
            branch.unlock()
 
352
        finally:
 
353
            branch.unlock()
 
354
        # The lock should be unlocked on disk.  Verify that with a new lock
 
355
        # instance.
 
356
        new_branch = branch.bzrdir.open_branch()
 
357
        # Calling lock_write now should work, rather than raise LockContention.
 
358
        new_branch.lock_write()
 
359
        new_branch.unlock()
 
360
 
 
361
    def test_leave_lock_in_place(self):
 
362
        branch = self.make_branch('b')
 
363
        # Lock the branch, then use leave_lock_in_place so that when we
 
364
        # unlock the branch the lock is still held on disk.
 
365
        token = branch.lock_write()
 
366
        try:
 
367
            if token is None:
 
368
                # This test does not apply, because this repository refuses lock
 
369
                # tokens.
 
370
                self.assertRaises(NotImplementedError,
 
371
                                  branch.leave_lock_in_place)
 
372
                return
 
373
            branch.leave_lock_in_place()
 
374
        finally:
 
375
            branch.unlock()
 
376
        # We should be unable to relock the repo.
 
377
        self.assertRaises(errors.LockContention, branch.lock_write)
 
378
 
 
379
    def test_dont_leave_lock_in_place(self):
 
380
        branch = self.make_branch('b')
 
381
        # Create a lock on disk.
 
382
        token = branch.lock_write()
 
383
        try:
 
384
            if token is None:
 
385
                # This test does not apply, because this branch refuses lock
 
386
                # tokens.
 
387
                self.assertRaises(NotImplementedError,
 
388
                                  branch.dont_leave_lock_in_place)
 
389
                return
 
390
            try:
 
391
                branch.leave_lock_in_place()
 
392
            except NotImplementedError:
 
393
                # This branch doesn't support this API.
 
394
                return
 
395
            branch.repository.leave_lock_in_place()
 
396
            repo_token = branch.repository.lock_write()
 
397
            branch.repository.unlock()
 
398
        finally:
 
399
            branch.unlock()
 
400
        # Reacquire the lock (with a different branch object) by using the
 
401
        # tokens.
 
402
        new_branch = branch.bzrdir.open_branch()
 
403
        # We have to explicitly lock the repository first.
 
404
        new_branch.repository.lock_write(token=repo_token)
 
405
        new_branch.lock_write(token=token)
 
406
        # Now we don't need our own repository lock anymore (the branch is
 
407
        # holding it for us).
 
408
        new_branch.repository.unlock()
 
409
        # Call dont_leave_lock_in_place, so that the lock will be released by
 
410
        # this instance, even though the lock wasn't originally acquired by it.
 
411
        new_branch.dont_leave_lock_in_place()
 
412
        new_branch.repository.dont_leave_lock_in_place()
 
413
        new_branch.unlock()
 
414
        # Now the branch (and repository) is unlocked.  Test this by locking it
 
415
        # without tokens.
 
416
        branch.lock_write()
 
417
        branch.unlock()
 
418
 
230
419
    def test_lock_read_then_unlock(self):
231
420
        # Calling lock_read then unlocking should work without errors.
232
421
        branch = self.make_branch('b')