43
43
def resetTimeout():
44
44
lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
45
45
self.addCleanup(resetTimeout)
46
lockdir._DEFAULT_TIMEOUT_SECONDS = 3
46
lockdir._DEFAULT_TIMEOUT_SECONDS = 0
48
48
def test_read_write(self):
49
49
self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
240
240
new_lockable.lock_write()
241
241
new_lockable.unlock()
243
def test_leave_in_place(self):
244
token = self.lockable.lock_write()
247
# This test does not apply, because this lockable refuses
250
self.lockable.leave_in_place()
252
self.lockable.unlock()
253
# At this point, the lock is still in place on disk
254
self.assertRaises(errors.LockContention, self.lockable.lock_write)
255
# But should be relockable with a token.
256
self.lockable.lock_write(token=token)
257
self.lockable.unlock()
259
def test_dont_leave_in_place(self):
260
token = self.lockable.lock_write()
263
# This test does not apply, because this lockable refuses
266
self.lockable.leave_in_place()
268
self.lockable.unlock()
269
# At this point, the lock is still in place on disk.
270
# Acquire the existing lock with the token, and ask that it is removed
271
# when this object unlocks, and unlock to trigger that removal.
272
new_lockable = self.get_lockable()
273
new_lockable.lock_write(token=token)
274
new_lockable.dont_leave_in_place()
275
new_lockable.unlock()
276
# At this point, the lock is no longer on disk, so we can lock it.
277
third_lockable = self.get_lockable()
278
third_lockable.lock_write()
279
third_lockable.unlock()
247
282
# This method of adapting tests to parameters is different to
305
340
"""LockableFile tests run with RemoteLockDir on a branch."""
308
super(TestLockableFiles_RemoteLockDir, self).setUp()
343
TestCaseWithSmartMedium.setUp(self)
344
_TestLockableFiles_mixin.setUp(self)
309
345
# can only get a RemoteLockDir with some RemoteObject...
310
346
# use a branch as thats what we want. These mixin tests test the end
311
347
# to end behaviour, so stubbing out the backend and simulating would
319
355
# getting a new lockable involves opening a new instance of the branch
320
356
branch = bzrlib.branch.Branch.open(self.get_url('foo'))
321
357
return branch.control_files
359
def test_lock_write_returns_None_refuses_token(self):
360
# this test is not relevant for RemoteBranchLockableFiles as remote
361
# locks are done directly from the remote branch object.
364
def test_lock_write_raises_on_token_mismatch(self):
365
# See test_lock_write_returns_None_refuses_token.
368
def test_lock_write_with_matching_token(self):
369
# See test_lock_write_returns_None_refuses_token.
372
def test_unlock_after_lock_write_with_token(self):
373
# See test_lock_write_returns_None_refuses_token.
376
def test_lock_write_with_token_fails_when_unlocked(self):
377
# See test_lock_write_returns_None_refuses_token.
380
def test_lock_write_reenter_with_token(self):
381
# See test_lock_write_returns_None_refuses_token.
384
def test_leave_in_place(self):
385
# See test_lock_write_returns_None_refuses_token.
388
def test_dont_leave_in_place(self):
389
# See test_lock_write_returns_None_refuses_token.