/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

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

  • Committer: Andrew Bennetts
  • Date: 2007-04-11 13:35:32 UTC
  • mto: (2018.5.146 hpss)
  • mto: This revision was merged to the branch mainline in revision 2414.
  • Revision ID: andrew.bennetts@canonical.com-20070411133532-u6x6edf3dmzamnaq
LockDir, Repository and Branch lock token changes from the hpss branch.

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