/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/test_lockable_files.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/test_lockable_files.py
36
36
class _TestLockableFiles_mixin(object):
37
37
 
38
38
    def setUp(self):
39
 
        # Reduce the default timeout, so that if tests fail, they will do so
40
 
        # reasonably quickly.
41
 
        orig_timeout = lockdir._DEFAULT_TIMEOUT_SECONDS
42
 
        def resetTimeout():
43
 
            lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
44
 
        self.addCleanup(resetTimeout)
45
 
        lockdir._DEFAULT_TIMEOUT_SECONDS = 3
 
39
        self.reduceLockdirTimeout()
46
40
 
47
41
    def test_read_write(self):
48
42
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
147
141
        finally:
148
142
            self.lockable.unlock()
149
143
 
 
144
    def test_lock_write_returns_token_when_given_token(self):
 
145
        token = self.lockable.lock_write()
 
146
        try:
 
147
            if token is None:
 
148
                # This test does not apply, because this lockable refuses
 
149
                # tokens.
 
150
                return
 
151
            new_lockable = self.get_lockable()
 
152
            token_from_new_lockable = new_lockable.lock_write(token=token)
 
153
            try:
 
154
                self.assertEqual(token, token_from_new_lockable)
 
155
            finally:
 
156
                new_lockable.unlock()
 
157
        finally:
 
158
            self.lockable.unlock()
 
159
 
150
160
    def test_lock_write_raises_on_token_mismatch(self):
151
161
        token = self.lockable.lock_write()
152
162
        try:
231
241
                # tokens.
232
242
                return
233
243
            # Relock with a token.
234
 
            self.lockable.lock_write(token=token)
235
 
            self.lockable.unlock()
 
244
            token_from_reentry = self.lockable.lock_write(token=token)
 
245
            try:
 
246
                self.assertEqual(token, token_from_reentry)
 
247
            finally:
 
248
                self.lockable.unlock()
236
249
        finally:
237
250
            self.lockable.unlock()
238
251
        # The lock should be unlocked on disk.  Verify that with a new lock
242
255
        new_lockable.lock_write()
243
256
        new_lockable.unlock()
244
257
 
 
258
    def test_leave_in_place(self):
 
259
        token = self.lockable.lock_write()
 
260
        try:
 
261
            if token is None:
 
262
                # This test does not apply, because this lockable refuses
 
263
                # tokens.
 
264
                return
 
265
            self.lockable.leave_in_place()
 
266
        finally:
 
267
            self.lockable.unlock()
 
268
        # At this point, the lock is still in place on disk
 
269
        self.assertRaises(errors.LockContention, self.lockable.lock_write)
 
270
        # But should be relockable with a token.
 
271
        self.lockable.lock_write(token=token)
 
272
        self.lockable.unlock()
 
273
 
 
274
    def test_dont_leave_in_place(self):
 
275
        token = self.lockable.lock_write()
 
276
        try:
 
277
            if token is None:
 
278
                # This test does not apply, because this lockable refuses
 
279
                # tokens.
 
280
                return
 
281
            self.lockable.leave_in_place()
 
282
        finally:
 
283
            self.lockable.unlock()
 
284
        # At this point, the lock is still in place on disk.
 
285
        # Acquire the existing lock with the token, and ask that it is removed
 
286
        # when this object unlocks, and unlock to trigger that removal.
 
287
        new_lockable = self.get_lockable()
 
288
        new_lockable.lock_write(token=token)
 
289
        new_lockable.dont_leave_in_place()
 
290
        new_lockable.unlock()
 
291
        # At this point, the lock is no longer on disk, so we can lock it.
 
292
        third_lockable = self.get_lockable()
 
293
        third_lockable.lock_write()
 
294
        third_lockable.unlock()
 
295
 
 
296
 
245
297
 
246
298
 
247
299