← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: Martin Pool
  • Date: 2006-02-21 22:46:09 UTC
  • mto: This revision was merged to the branch mainline in revision 1569.
  • Revision ID: mbp@sourcefrog.net-20060221224609-a7f433d0488c080d
Confirm that only the intended holder of a lock was broken.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
19
19
from threading import Thread
20
20
import time
21
21
 
22
 
from bzrlib.errors import LockContention, LockError, UnlockableTransport, \
 
22
from bzrlib.errors import (
 
23
        LockBreakMismatch,
 
24
        LockContention, LockError, UnlockableTransport,
23
25
        LockNotHeld, LockBroken
 
26
        )
24
27
from bzrlib.lockdir import LockDir
25
28
from bzrlib.tests import TestCaseInTempDir, TestCaseWithTransport
26
29
 
225
228
        del lf1
226
229
        # someone else sees it's still locked
227
230
        lf2 = LockDir(t, 'test_lock')
228
 
        self.assertTrue(lf2.peek())
229
 
        lf2.force_break()
 
231
        holder_info = lf2.peek()
 
232
        self.assertTrue(holder_info)
 
233
        lf2.force_break(holder_info)
230
234
        # now we should be able to take it
231
235
        lf2.attempt_lock()
232
236
        lf2.confirm()
242
246
        # in the interim the lock is released
243
247
        lf1.unlock()
244
248
        # break should succeed
245
 
        lf2.force_break()
 
249
        lf2.force_break(holder_info)
246
250
        # now we should be able to take it
247
251
        lf2.attempt_lock()
248
252
        lf2.confirm()
249
253
 
 
254
    def test_45_break_mismatch(self):
 
255
        """Lock break races with someone else acquiring it"""
 
256
        t = self.get_transport()
 
257
        lf1 = LockDir(t, 'test_lock')
 
258
        lf1.attempt_lock()
 
259
        # someone else sees it's still locked
 
260
        lf2 = LockDir(t, 'test_lock')
 
261
        holder_info = lf2.peek()
 
262
        # in the interim the lock is released
 
263
        lf1.unlock()
 
264
        lf3 = LockDir(t, 'test_lock')
 
265
        lf3.attempt_lock()
 
266
        # break should now *fail*
 
267
        self.assertRaises(LockBreakMismatch, lf2.force_break,
 
268
                          holder_info)
 
269
        lf3.unlock()