1
# Copyright (C) 2006-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for LockDir"""
20
from threading import Thread, Lock
32
from bzrlib.errors import (
39
from bzrlib.lockdir import LockDir
40
from bzrlib.tests import (
42
TestCaseWithTransport,
44
from bzrlib.trace import note
46
# These tests sometimes use threads to test the behaviour of lock files with
47
# concurrent actors. This is not a typical (or necessarily supported) use;
48
# they're really meant for guarding between processes.
50
# These tests are run on the default transport provided by the test framework
51
# (typically a local disk transport). That can be changed by the --transport
52
# option to bzr selftest. The required properties of the transport
53
# implementation are tested separately. (The main requirement is just that
54
# they don't allow overwriting nonempty directories.)
56
class TestLockDir(TestCaseWithTransport):
57
"""Test LockDir operations"""
59
def logging_report_function(self, fmt, *args):
60
self._logged_reports.append((fmt, args))
62
def setup_log_reporter(self, lock_dir):
63
self._logged_reports = []
64
lock_dir._report_function = self.logging_report_function
66
def test_00_lock_creation(self):
67
"""Creation of lock file on a transport"""
68
t = self.get_transport()
69
lf = LockDir(t, 'test_lock')
70
self.assertFalse(lf.is_held)
72
def test_01_lock_repr(self):
73
"""Lock string representation"""
74
lf = LockDir(self.get_transport(), 'test_lock')
76
self.assertContainsRe(r, r'^LockDir\(.*/test_lock\)$')
78
def test_02_unlocked_peek(self):
79
lf = LockDir(self.get_transport(), 'test_lock')
80
self.assertEqual(lf.peek(), None)
83
return LockDir(self.get_transport(), 'test_lock')
85
def test_unlock_after_break_raises(self):
90
ld2.force_break(ld2.peek())
91
self.assertRaises(LockBroken, ld.unlock)
93
def test_03_readonly_peek(self):
94
lf = LockDir(self.get_readonly_transport(), 'test_lock')
95
self.assertEqual(lf.peek(), None)
97
def test_10_lock_uncontested(self):
98
"""Acquire and release a lock"""
99
t = self.get_transport()
100
lf = LockDir(t, 'test_lock')
104
self.assertTrue(lf.is_held)
107
self.assertFalse(lf.is_held)
109
def test_11_create_readonly_transport(self):
110
"""Fail to create lock on readonly transport"""
111
t = self.get_readonly_transport()
112
lf = LockDir(t, 'test_lock')
113
self.assertRaises(LockFailed, lf.create)
115
def test_12_lock_readonly_transport(self):
116
"""Fail to lock on readonly transport"""
117
lf = LockDir(self.get_transport(), 'test_lock')
119
lf = LockDir(self.get_readonly_transport(), 'test_lock')
120
self.assertRaises(LockFailed, lf.attempt_lock)
122
def test_20_lock_contested(self):
123
"""Contention to get a lock"""
124
t = self.get_transport()
125
lf1 = LockDir(t, 'test_lock')
128
lf2 = LockDir(t, 'test_lock')
130
# locking is between LockDir instances; aliases within
131
# a single process are not detected
133
self.fail('Failed to detect lock collision')
134
except LockContention, e:
135
self.assertEqual(e.lock, lf2)
136
self.assertContainsRe(str(e),
137
r'^Could not acquire.*test_lock.*$')
140
def test_20_lock_peek(self):
141
"""Peek at the state of a lock"""
142
t = self.get_transport()
143
lf1 = LockDir(t, 'test_lock')
146
self.addCleanup(lf1.unlock)
147
# lock is held, should get some info on it
149
self.assertEqual(set(info1.keys()),
150
set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
151
# should get the same info if we look at it through a different
153
info2 = LockDir(t, 'test_lock').peek()
154
self.assertEqual(info1, info2)
155
# locks which are never used should be not-held
156
self.assertEqual(LockDir(t, 'other_lock').peek(), None)
158
def test_21_peek_readonly(self):
159
"""Peek over a readonly transport"""
160
t = self.get_transport()
161
lf1 = LockDir(t, 'test_lock')
163
lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
164
self.assertEqual(lf2.peek(), None)
166
self.addCleanup(lf1.unlock)
168
self.assertTrue(info2)
169
self.assertEqual(info2['nonce'], lf1.nonce)
171
def test_30_lock_wait_fail(self):
172
"""Wait on a lock, then fail
174
We ask to wait up to 400ms; this should fail within at most one
175
second. (Longer times are more realistic but we don't want the test
176
suite to take too long, and this should do for now.)
178
t = self.get_transport()
179
lf1 = LockDir(t, 'test_lock')
181
lf2 = LockDir(t, 'test_lock')
182
self.setup_log_reporter(lf2)
186
self.assertRaises(LockContention, lf2.wait_lock,
187
timeout=0.4, poll=0.1)
189
# it should only take about 0.4 seconds, but we allow more time in
190
# case the machine is heavily loaded
191
self.assertTrue(after - before <= 8.0,
192
"took %f seconds to detect lock contention" % (after - before))
195
self.assertEqual(1, len(self._logged_reports))
196
self.assertEqual(self._logged_reports[0][0],
197
'%s lock %s held by %s\n'
198
'at %s [process #%s], acquired %s.\n'
199
'Will continue to try until %s, unless '
200
'you press Ctrl-C.\n'
201
'See "bzr help break-lock" for more.')
202
start, lock_url, user, hostname, pid, time_ago, deadline_str = \
203
self._logged_reports[0][1]
204
self.assertEqual(start, u'Unable to obtain')
205
self.assertEqual(user, u'jrandom@example.com')
207
self.assertContainsRe(pid, r'\d+')
208
self.assertContainsRe(time_ago, r'.* ago')
209
self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
211
def test_31_lock_wait_easy(self):
212
"""Succeed when waiting on a lock with no contention.
214
t = self.get_transport()
215
lf1 = LockDir(t, 'test_lock')
217
self.setup_log_reporter(lf1)
220
lf1.wait_lock(timeout=0.4, poll=0.1)
222
self.assertTrue(after - before <= 1.0)
225
self.assertEqual([], self._logged_reports)
227
def test_32_lock_wait_succeed(self):
228
"""Succeed when trying to acquire a lock that gets released
230
One thread holds on a lock and then releases it; another
233
# This test sometimes fails like this:
234
# Traceback (most recent call last):
236
# File "/home/pqm/bzr-pqm-workdir/home/+trunk/bzrlib/tests/
237
# test_lockdir.py", line 247, in test_32_lock_wait_succeed
238
# self.assertEqual(1, len(self._logged_reports))
239
# AssertionError: not equal:
242
raise tests.TestSkipped("Test fails intermittently")
243
t = self.get_transport()
244
lf1 = LockDir(t, 'test_lock')
248
def wait_and_unlock():
251
unlocker = Thread(target=wait_and_unlock)
254
lf2 = LockDir(t, 'test_lock')
255
self.setup_log_reporter(lf2)
258
lf2.wait_lock(timeout=0.4, poll=0.1)
260
self.assertTrue(after - before <= 1.0)
264
# There should be only 1 report, even though it should have to
266
lock_base = lf2.transport.abspath(lf2.path)
267
self.assertEqual(1, len(self._logged_reports))
268
self.assertEqual('%s %s\n'
270
'Will continue to try until %s\n',
271
self._logged_reports[0][0])
272
args = self._logged_reports[0][1]
273
self.assertEqual('Unable to obtain', args[0])
274
self.assertEqual('lock %s' % (lock_base,), args[1])
275
self.assertStartsWith(args[2], 'held by ')
276
self.assertStartsWith(args[3], 'locked ')
277
self.assertEndsWith(args[3], ' ago')
278
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
280
def test_34_lock_write_waits(self):
281
"""LockDir.lock_write() will wait for the lock."""
282
# the test suite sets the default to 0 to make deadlocks fail fast.
283
# change it for this test, as we want to try a manual deadlock.
284
raise tests.TestSkipped('Timing-sensitive test')
285
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300
286
t = self.get_transport()
287
lf1 = LockDir(t, 'test_lock')
291
def wait_and_unlock():
294
unlocker = Thread(target=wait_and_unlock)
297
lf2 = LockDir(t, 'test_lock')
298
self.setup_log_reporter(lf2)
306
# There should be only 1 report, even though it should have to
308
lock_base = lf2.transport.abspath(lf2.path)
309
self.assertEqual(1, len(self._logged_reports))
310
self.assertEqual('%s %s\n'
312
'Will continue to try until %s\n',
313
self._logged_reports[0][0])
314
args = self._logged_reports[0][1]
315
self.assertEqual('Unable to obtain', args[0])
316
self.assertEqual('lock %s' % (lock_base,), args[1])
317
self.assertStartsWith(args[2], 'held by ')
318
self.assertStartsWith(args[3], 'locked ')
319
self.assertEndsWith(args[3], ' ago')
320
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
322
def test_35_wait_lock_changing(self):
323
"""LockDir.wait_lock() will report if the lock changes underneath.
325
This is the stages we want to happen:
327
0) Synchronization locks are created and locked.
328
1) Lock1 obtains the lockdir, and releases the 'check' lock.
329
2) Lock2 grabs the 'check' lock, and checks the lockdir.
330
It sees the lockdir is already acquired, reports the fact,
331
and unsets the 'checked' lock.
332
3) Thread1 blocks on acquiring the 'checked' lock, and then tells
333
Lock1 to release and acquire the lockdir. This resets the 'check'
335
4) Lock2 acquires the 'check' lock, and checks again. It notices
336
that the holder of the lock has changed, and so reports a new
338
5) Thread1 blocks on the 'checked' lock, this time, it completely
339
unlocks the lockdir, allowing Lock2 to acquire the lock.
342
raise tests.KnownFailure(
343
"timing dependency in lock tests (#213182)")
345
wait_to_check_lock = Lock()
346
wait_until_checked_lock = Lock()
348
wait_to_check_lock.acquire()
349
wait_until_checked_lock.acquire()
350
note('locked check and checked locks')
352
class LockDir1(LockDir):
353
"""Use the synchronization points for the first lock."""
355
def attempt_lock(self):
356
# Once we have acquired the lock, it is okay for
357
# the other lock to check it
359
return super(LockDir1, self).attempt_lock()
361
note('lock1: releasing check lock')
362
wait_to_check_lock.release()
364
class LockDir2(LockDir):
365
"""Use the synchronization points for the second lock."""
367
def attempt_lock(self):
368
note('lock2: waiting for check lock')
369
wait_to_check_lock.acquire()
370
note('lock2: acquired check lock')
372
return super(LockDir2, self).attempt_lock()
374
note('lock2: releasing checked lock')
375
wait_until_checked_lock.release()
377
t = self.get_transport()
378
lf1 = LockDir1(t, 'test_lock')
381
lf2 = LockDir2(t, 'test_lock')
382
self.setup_log_reporter(lf2)
384
def wait_and_switch():
386
# Block until lock2 has had a chance to check
387
note('lock1: waiting 1 for checked lock')
388
wait_until_checked_lock.acquire()
389
note('lock1: acquired for checked lock')
390
note('lock1: released lockdir')
392
note('lock1: acquiring lockdir')
393
# Create a new nonce, so the lock looks different.
394
lf1.nonce = osutils.rand_chars(20)
396
note('lock1: acquired lockdir')
398
# Block until lock2 has peeked again
399
note('lock1: waiting 2 for checked lock')
400
wait_until_checked_lock.acquire()
401
note('lock1: acquired for checked lock')
402
# Now unlock, and let lock 2 grab the lock
404
wait_to_check_lock.release()
406
unlocker = Thread(target=wait_and_switch)
409
# Wait and play against the other thread
410
lf2.wait_lock(timeout=20.0, poll=0.01)
415
# There should be 2 reports, because the lock changed
416
lock_base = lf2.transport.abspath(lf2.path)
417
self.assertEqual(2, len(self._logged_reports))
418
lock_url = lf2.transport.abspath(lf2.path)
419
self.assertEqual('%s %s\n'
421
'Will continue to try until %s, unless '
422
'you press Ctrl-C.\n'
423
'See "bzr help break-lock" for more.',
424
self._logged_reports[0][0])
425
args = self._logged_reports[0][1]
426
self.assertEqual('Unable to obtain', args[0])
427
self.assertEqual('lock %s' % (lock_base,), args[1])
428
self.assertStartsWith(args[2], 'held by ')
429
self.assertStartsWith(args[3], 'locked ')
430
self.assertEndsWith(args[3], ' ago')
431
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
433
self.assertEqual('%s %s\n'
435
'Will continue to try until %s, unless '
436
'you press Ctrl-C.\n'
437
'See "bzr help break-lock" for more.',
438
self._logged_reports[1][0])
439
args = self._logged_reports[1][1]
440
self.assertEqual('Lock owner changed for', args[0])
441
self.assertEqual('lock %s' % (lock_base,), args[1])
442
self.assertStartsWith(args[2], 'held by ')
443
self.assertStartsWith(args[3], 'locked ')
444
self.assertEndsWith(args[3], ' ago')
445
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
447
def test_40_confirm_easy(self):
448
"""Confirm a lock that's already held"""
449
t = self.get_transport()
450
lf1 = LockDir(t, 'test_lock')
453
self.addCleanup(lf1.unlock)
456
def test_41_confirm_not_held(self):
457
"""Confirm a lock that's already held"""
458
t = self.get_transport()
459
lf1 = LockDir(t, 'test_lock')
461
self.assertRaises(LockNotHeld, lf1.confirm)
463
def test_42_confirm_broken_manually(self):
464
"""Confirm a lock broken by hand"""
465
t = self.get_transport()
466
lf1 = LockDir(t, 'test_lock')
469
t.move('test_lock', 'lock_gone_now')
470
self.assertRaises(LockBroken, lf1.confirm)
472
t.move('lock_gone_now', 'test_lock')
475
def test_43_break(self):
476
"""Break a lock whose caller has forgotten it"""
477
t = self.get_transport()
478
lf1 = LockDir(t, 'test_lock')
481
# we incorrectly discard the lock object without unlocking it
483
# someone else sees it's still locked
484
lf2 = LockDir(t, 'test_lock')
485
holder_info = lf2.peek()
486
self.assertTrue(holder_info)
487
lf2.force_break(holder_info)
488
# now we should be able to take it
490
self.addCleanup(lf2.unlock)
493
def test_44_break_already_released(self):
494
"""Lock break races with regular release"""
495
t = self.get_transport()
496
lf1 = LockDir(t, 'test_lock')
499
# someone else sees it's still locked
500
lf2 = LockDir(t, 'test_lock')
501
holder_info = lf2.peek()
502
# in the interim the lock is released
504
# break should succeed
505
lf2.force_break(holder_info)
506
# now we should be able to take it
508
self.addCleanup(lf2.unlock)
511
def test_45_break_mismatch(self):
512
"""Lock break races with someone else acquiring it"""
513
t = self.get_transport()
514
lf1 = LockDir(t, 'test_lock')
517
# someone else sees it's still locked
518
lf2 = LockDir(t, 'test_lock')
519
holder_info = lf2.peek()
520
# in the interim the lock is released
522
lf3 = LockDir(t, 'test_lock')
524
# break should now *fail*
525
self.assertRaises(LockBreakMismatch, lf2.force_break,
529
def test_46_fake_read_lock(self):
530
t = self.get_transport()
531
lf1 = LockDir(t, 'test_lock')
536
def test_50_lockdir_representation(self):
537
"""Check the on-disk representation of LockDirs is as expected.
539
There should always be a top-level directory named by the lock.
540
When the lock is held, there should be a lockname/held directory
541
containing an info file.
543
t = self.get_transport()
544
lf1 = LockDir(t, 'test_lock')
546
self.assertTrue(t.has('test_lock'))
548
self.assertTrue(t.has('test_lock/held/info'))
550
self.assertFalse(t.has('test_lock/held/info'))
552
def test_break_lock(self):
553
# the ui based break_lock routine should Just Work (tm)
554
ld1 = self.get_lock()
555
ld2 = self.get_lock()
558
# do this without IO redirection to ensure it doesn't prompt.
559
self.assertRaises(AssertionError, ld1.break_lock)
560
orig_factory = bzrlib.ui.ui_factory
561
bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
564
self.assertRaises(LockBroken, ld1.unlock)
566
bzrlib.ui.ui_factory = orig_factory
568
def test_break_lock_corrupt_info(self):
569
"""break_lock works even if the info file is corrupt (and tells the UI
573
ld2 = self.get_lock()
576
ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
577
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
580
def get_boolean(self, prompt):
581
self.prompts.append(('boolean', prompt))
583
ui = LoggingUIFactory()
584
self.overrideAttr(bzrlib.ui, 'ui_factory', ui)
586
self.assertLength(1, ui.prompts)
587
self.assertEqual('boolean', ui.prompts[0][0])
588
self.assertStartsWith(ui.prompts[0][1], 'Break (corrupt LockDir')
589
self.assertRaises(LockBroken, ld.unlock)
591
def test_break_lock_missing_info(self):
592
"""break_lock works even if the info file is missing (and tells the UI
596
ld2 = self.get_lock()
599
ld.transport.delete('test_lock/held/info')
600
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
603
def get_boolean(self, prompt):
604
self.prompts.append(('boolean', prompt))
606
ui = LoggingUIFactory()
607
orig_factory = bzrlib.ui.ui_factory
608
bzrlib.ui.ui_factory = ui
611
self.assertRaises(LockBroken, ld.unlock)
612
self.assertLength(0, ui.prompts)
614
bzrlib.ui.ui_factory = orig_factory
615
# Suppress warnings due to ld not being unlocked
616
# XXX: if lock_broken hook was invoked in this case, this hack would
617
# not be necessary. - Andrew Bennetts, 2010-09-06.
618
del self._lock_actions[:]
620
def test_create_missing_base_directory(self):
621
"""If LockDir.path doesn't exist, it can be created
623
Some people manually remove the entire lock/ directory trying
624
to unlock a stuck repository/branch/etc. Rather than failing
625
after that, just create the lock directory when needed.
627
t = self.get_transport()
628
lf1 = LockDir(t, 'test_lock')
631
self.failUnless(t.has('test_lock'))
634
self.failIf(t.has('test_lock'))
636
# This will create 'test_lock' if it needs to
638
self.failUnless(t.has('test_lock'))
639
self.failUnless(t.has('test_lock/held/info'))
642
self.failIf(t.has('test_lock/held/info'))
644
def test__format_lock_info(self):
645
ld1 = self.get_lock()
649
info_list = ld1._format_lock_info(ld1.peek())
652
self.assertEqual(info_list[0], u'jrandom@example.com')
653
# info_list[1] is hostname. we skip this.
654
self.assertContainsRe(info_list[2], '^\d+$') # pid
655
self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
657
def test_lock_without_email(self):
658
global_config = config.GlobalConfig()
659
# Intentionally has no email address
660
global_config.set_user_option('email', 'User Identity')
661
ld1 = self.get_lock()
666
def test_lock_permission(self):
667
self.requireFeature(features.not_running_as_root)
668
if not osutils.supports_posix_readonly():
669
raise tests.TestSkipped('Cannot induce a permission failure')
670
ld1 = self.get_lock()
671
lock_path = ld1.transport.local_abspath('test_lock')
673
osutils.make_readonly(lock_path)
674
self.assertRaises(errors.LockFailed, ld1.attempt_lock)
676
def test_lock_by_token(self):
677
ld1 = self.get_lock()
678
token = ld1.lock_write()
679
self.addCleanup(ld1.unlock)
680
self.assertNotEqual(None, token)
681
ld2 = self.get_lock()
682
t2 = ld2.lock_write(token)
683
self.addCleanup(ld2.unlock)
684
self.assertEqual(token, t2)
686
def test_lock_with_buggy_rename(self):
687
# test that lock acquisition handles servers which pretend they
688
# renamed correctly but that actually fail
689
t = transport.get_transport('brokenrename+' + self.get_url())
690
ld1 = LockDir(t, 'test_lock')
693
ld2 = LockDir(t, 'test_lock')
694
# we should fail to lock
695
e = self.assertRaises(errors.LockContention, ld2.attempt_lock)
696
# now the original caller should succeed in unlocking
698
# and there should be nothing left over
699
self.assertEquals([], t.list_dir('test_lock'))
701
def test_failed_lock_leaves_no_trash(self):
702
# if we fail to acquire the lock, we don't leave pending directories
703
# behind -- https://bugs.launchpad.net/bzr/+bug/109169
704
ld1 = self.get_lock()
705
ld2 = self.get_lock()
706
# should be nothing before we start
708
t = self.get_transport().clone('test_lock')
710
self.assertEquals(a, t.list_dir('.'))
712
# when held, that's all we see
714
self.addCleanup(ld1.unlock)
716
# second guy should fail
717
self.assertRaises(errors.LockContention, ld2.attempt_lock)
721
def test_no_lockdir_info(self):
722
"""We can cope with empty info files."""
723
# This seems like a fairly common failure case - see
724
# <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
725
# Processes are often interrupted after opening the file
726
# before the actual contents are committed.
727
t = self.get_transport()
729
t.mkdir('test_lock/held')
730
t.put_bytes('test_lock/held/info', '')
731
lf = LockDir(t, 'test_lock')
733
formatted_info = lf._format_lock_info(info)
735
['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
738
def test_corrupt_lockdir_info(self):
739
"""We can cope with corrupt (and thus unparseable) info files."""
740
# This seems like a fairly common failure case too - see
741
# <https://bugs.launchpad.net/bzr/+bug/619872> for instance.
742
# In particular some systems tend to fill recently created files with
743
# nul bytes after recovering from a system crash.
744
t = self.get_transport()
746
t.mkdir('test_lock/held')
747
t.put_bytes('test_lock/held/info', '\0')
748
lf = LockDir(t, 'test_lock')
749
self.assertRaises(errors.LockCorrupt, lf.peek)
750
# Currently attempt_lock gives LockContention, but LockCorrupt would be
751
# a reasonable result too.
753
(errors.LockCorrupt, errors.LockContention), lf.attempt_lock)
754
self.assertRaises(errors.LockCorrupt, lf.validate_token, 'fake token')
756
def test_missing_lockdir_info(self):
757
"""We can cope with absent info files."""
758
t = self.get_transport()
760
t.mkdir('test_lock/held')
761
lf = LockDir(t, 'test_lock')
762
# In this case we expect the 'not held' result from peek, because peek
763
# cannot be expected to notice that there is a 'held' directory with no
765
self.assertEqual(None, lf.peek())
766
# And lock/unlock may work or give LockContention (but not any other
770
except LockContention:
771
# LockContention is ok, and expected on Windows
774
# no error is ok, and expected on POSIX (because POSIX allows
775
# os.rename over an empty directory).
777
# Currently raises TokenMismatch, but LockCorrupt would be reasonable
780
(errors.TokenMismatch, errors.LockCorrupt),
781
lf.validate_token, 'fake token')
784
class TestLockDirHooks(TestCaseWithTransport):
787
super(TestLockDirHooks, self).setUp()
791
return LockDir(self.get_transport(), 'test_lock')
793
def record_hook(self, result):
794
self._calls.append(result)
796
def test_LockDir_acquired_success(self):
797
# the LockDir.lock_acquired hook fires when a lock is acquired.
798
LockDir.hooks.install_named_hook('lock_acquired',
799
self.record_hook, 'record_hook')
802
self.assertEqual([], self._calls)
803
result = ld.attempt_lock()
804
lock_path = ld.transport.abspath(ld.path)
805
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
807
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
809
def test_LockDir_acquired_fail(self):
810
# the LockDir.lock_acquired hook does not fire on failure.
813
ld2 = self.get_lock()
815
# install a lock hook now, when the disk lock is locked
816
LockDir.hooks.install_named_hook('lock_acquired',
817
self.record_hook, 'record_hook')
818
self.assertRaises(errors.LockContention, ld.attempt_lock)
819
self.assertEqual([], self._calls)
821
self.assertEqual([], self._calls)
823
def test_LockDir_released_success(self):
824
# the LockDir.lock_released hook fires when a lock is acquired.
825
LockDir.hooks.install_named_hook('lock_released',
826
self.record_hook, 'record_hook')
829
self.assertEqual([], self._calls)
830
result = ld.attempt_lock()
831
self.assertEqual([], self._calls)
833
lock_path = ld.transport.abspath(ld.path)
834
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
836
def test_LockDir_released_fail(self):
837
# the LockDir.lock_released hook does not fire on failure.
840
ld2 = self.get_lock()
842
ld2.force_break(ld2.peek())
843
LockDir.hooks.install_named_hook('lock_released',
844
self.record_hook, 'record_hook')
845
self.assertRaises(LockBroken, ld.unlock)
846
self.assertEqual([], self._calls)
848
def test_LockDir_broken_success(self):
849
# the LockDir.lock_broken hook fires when a lock is broken.
852
ld2 = self.get_lock()
853
result = ld.attempt_lock()
854
LockDir.hooks.install_named_hook('lock_broken',
855
self.record_hook, 'record_hook')
856
ld2.force_break(ld2.peek())
857
lock_path = ld.transport.abspath(ld.path)
858
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
860
def test_LockDir_broken_failure(self):
861
# the LockDir.lock_broken hook does not fires when a lock is already
865
ld2 = self.get_lock()
866
result = ld.attempt_lock()
867
holder_info = ld2.peek()
869
LockDir.hooks.install_named_hook('lock_broken',
870
self.record_hook, 'record_hook')
871
ld2.force_break(holder_info)
872
lock_path = ld.transport.abspath(ld.path)
873
self.assertEqual([], self._calls)
added
removed
bzrlib/tests/test_lockdir.py
