1
# Copyright (C) 2006-2010 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"""
19
from cStringIO import StringIO
21
from threading import Thread, Lock
33
from bzrlib.errors import (
41
from bzrlib.lockdir import LockDir
42
from bzrlib.tests import (features, TestCaseWithTransport)
43
from bzrlib.trace import note
45
# These tests sometimes use threads to test the behaviour of lock files with
46
# concurrent actors. This is not a typical (or necessarily supported) use;
47
# they're really meant for guarding between processes.
49
# These tests are run on the default transport provided by the test framework
50
# (typically a local disk transport). That can be changed by the --transport
51
# option to bzr selftest. The required properties of the transport
52
# implementation are tested separately. (The main requirement is just that
53
# they don't allow overwriting nonempty directories.)
55
class TestLockDir(TestCaseWithTransport):
56
"""Test LockDir operations"""
58
def logging_report_function(self, fmt, *args):
59
self._logged_reports.append((fmt, args))
61
def setup_log_reporter(self, lock_dir):
62
self._logged_reports = []
63
lock_dir._report_function = self.logging_report_function
65
def test_00_lock_creation(self):
66
"""Creation of lock file on a transport"""
67
t = self.get_transport()
68
lf = LockDir(t, 'test_lock')
69
self.assertFalse(lf.is_held)
71
def test_01_lock_repr(self):
72
"""Lock string representation"""
73
lf = LockDir(self.get_transport(), 'test_lock')
75
self.assertContainsRe(r, r'^LockDir\(.*/test_lock\)$')
77
def test_02_unlocked_peek(self):
78
lf = LockDir(self.get_transport(), 'test_lock')
79
self.assertEqual(lf.peek(), None)
82
return LockDir(self.get_transport(), 'test_lock')
84
def test_unlock_after_break_raises(self):
89
ld2.force_break(ld2.peek())
90
self.assertRaises(LockBroken, ld.unlock)
92
def test_03_readonly_peek(self):
93
lf = LockDir(self.get_readonly_transport(), 'test_lock')
94
self.assertEqual(lf.peek(), None)
96
def test_10_lock_uncontested(self):
97
"""Acquire and release a lock"""
98
t = self.get_transport()
99
lf = LockDir(t, 'test_lock')
103
self.assertTrue(lf.is_held)
106
self.assertFalse(lf.is_held)
108
def test_11_create_readonly_transport(self):
109
"""Fail to create lock on readonly transport"""
110
t = self.get_readonly_transport()
111
lf = LockDir(t, 'test_lock')
112
self.assertRaises(LockFailed, lf.create)
114
def test_12_lock_readonly_transport(self):
115
"""Fail to lock on readonly transport"""
116
lf = LockDir(self.get_transport(), 'test_lock')
118
lf = LockDir(self.get_readonly_transport(), 'test_lock')
119
self.assertRaises(LockFailed, lf.attempt_lock)
121
def test_20_lock_contested(self):
122
"""Contention to get a lock"""
123
t = self.get_transport()
124
lf1 = LockDir(t, 'test_lock')
127
lf2 = LockDir(t, 'test_lock')
129
# locking is between LockDir instances; aliases within
130
# a single process are not detected
132
self.fail('Failed to detect lock collision')
133
except LockContention, e:
134
self.assertEqual(e.lock, lf2)
135
self.assertContainsRe(str(e),
136
r'^Could not acquire.*test_lock.*$')
139
def test_20_lock_peek(self):
140
"""Peek at the state of a lock"""
141
t = self.get_transport()
142
lf1 = LockDir(t, 'test_lock')
145
self.addCleanup(lf1.unlock)
146
# lock is held, should get some info on it
148
self.assertEqual(set(info1.keys()),
149
set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
150
# should get the same info if we look at it through a different
152
info2 = LockDir(t, 'test_lock').peek()
153
self.assertEqual(info1, info2)
154
# locks which are never used should be not-held
155
self.assertEqual(LockDir(t, 'other_lock').peek(), None)
157
def test_21_peek_readonly(self):
158
"""Peek over a readonly transport"""
159
t = self.get_transport()
160
lf1 = LockDir(t, 'test_lock')
162
lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
163
self.assertEqual(lf2.peek(), None)
165
self.addCleanup(lf1.unlock)
167
self.assertTrue(info2)
168
self.assertEqual(info2['nonce'], lf1.nonce)
170
def test_30_lock_wait_fail(self):
171
"""Wait on a lock, then fail
173
We ask to wait up to 400ms; this should fail within at most one
174
second. (Longer times are more realistic but we don't want the test
175
suite to take too long, and this should do for now.)
177
t = self.get_transport()
178
lf1 = LockDir(t, 'test_lock')
180
lf2 = LockDir(t, 'test_lock')
181
self.setup_log_reporter(lf2)
185
self.assertRaises(LockContention, lf2.wait_lock,
186
timeout=0.4, poll=0.1)
188
# it should only take about 0.4 seconds, but we allow more time in
189
# case the machine is heavily loaded
190
self.assertTrue(after - before <= 8.0,
191
"took %f seconds to detect lock contention" % (after - before))
194
self.assertEqual(1, len(self._logged_reports))
195
self.assertEqual(self._logged_reports[0][0],
196
'%s lock %s held by %s\n'
197
'at %s [process #%s], acquired %s.\n'
198
'Will continue to try until %s, unless '
199
'you press Ctrl-C.\n'
200
'See "bzr help break-lock" for more.')
201
start, lock_url, user, hostname, pid, time_ago, deadline_str = \
202
self._logged_reports[0][1]
203
self.assertEqual(start, u'Unable to obtain')
204
self.assertEqual(user, u'jrandom@example.com')
206
self.assertContainsRe(pid, r'\d+')
207
self.assertContainsRe(time_ago, r'.* ago')
208
self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
210
def test_31_lock_wait_easy(self):
211
"""Succeed when waiting on a lock with no contention.
213
t = self.get_transport()
214
lf1 = LockDir(t, 'test_lock')
216
self.setup_log_reporter(lf1)
219
lf1.wait_lock(timeout=0.4, poll=0.1)
221
self.assertTrue(after - before <= 1.0)
224
self.assertEqual([], self._logged_reports)
226
def test_32_lock_wait_succeed(self):
227
"""Succeed when trying to acquire a lock that gets released
229
One thread holds on a lock and then releases it; another
232
# This test sometimes fails like this:
233
# Traceback (most recent call last):
235
# File "/home/pqm/bzr-pqm-workdir/home/+trunk/bzrlib/tests/
236
# test_lockdir.py", line 247, in test_32_lock_wait_succeed
237
# self.assertEqual(1, len(self._logged_reports))
238
# AssertionError: not equal:
241
raise tests.TestSkipped("Test fails intermittently")
242
t = self.get_transport()
243
lf1 = LockDir(t, 'test_lock')
247
def wait_and_unlock():
250
unlocker = Thread(target=wait_and_unlock)
253
lf2 = LockDir(t, 'test_lock')
254
self.setup_log_reporter(lf2)
257
lf2.wait_lock(timeout=0.4, poll=0.1)
259
self.assertTrue(after - before <= 1.0)
263
# There should be only 1 report, even though it should have to
265
lock_base = lf2.transport.abspath(lf2.path)
266
self.assertEqual(1, len(self._logged_reports))
267
self.assertEqual('%s %s\n'
269
'Will continue to try until %s\n',
270
self._logged_reports[0][0])
271
args = self._logged_reports[0][1]
272
self.assertEqual('Unable to obtain', args[0])
273
self.assertEqual('lock %s' % (lock_base,), args[1])
274
self.assertStartsWith(args[2], 'held by ')
275
self.assertStartsWith(args[3], 'locked ')
276
self.assertEndsWith(args[3], ' ago')
277
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
279
def test_34_lock_write_waits(self):
280
"""LockDir.lock_write() will wait for the lock."""
281
# the test suite sets the default to 0 to make deadlocks fail fast.
282
# change it for this test, as we want to try a manual deadlock.
283
raise tests.TestSkipped('Timing-sensitive test')
284
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300
285
t = self.get_transport()
286
lf1 = LockDir(t, 'test_lock')
290
def wait_and_unlock():
293
unlocker = Thread(target=wait_and_unlock)
296
lf2 = LockDir(t, 'test_lock')
297
self.setup_log_reporter(lf2)
305
# There should be only 1 report, even though it should have to
307
lock_base = lf2.transport.abspath(lf2.path)
308
self.assertEqual(1, len(self._logged_reports))
309
self.assertEqual('%s %s\n'
311
'Will continue to try until %s\n',
312
self._logged_reports[0][0])
313
args = self._logged_reports[0][1]
314
self.assertEqual('Unable to obtain', args[0])
315
self.assertEqual('lock %s' % (lock_base,), args[1])
316
self.assertStartsWith(args[2], 'held by ')
317
self.assertStartsWith(args[3], 'locked ')
318
self.assertEndsWith(args[3], ' ago')
319
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
321
def test_35_wait_lock_changing(self):
322
"""LockDir.wait_lock() will report if the lock changes underneath.
324
This is the stages we want to happen:
326
0) Synchronization locks are created and locked.
327
1) Lock1 obtains the lockdir, and releases the 'check' lock.
328
2) Lock2 grabs the 'check' lock, and checks the lockdir.
329
It sees the lockdir is already acquired, reports the fact,
330
and unsets the 'checked' lock.
331
3) Thread1 blocks on acquiring the 'checked' lock, and then tells
332
Lock1 to release and acquire the lockdir. This resets the 'check'
334
4) Lock2 acquires the 'check' lock, and checks again. It notices
335
that the holder of the lock has changed, and so reports a new
337
5) Thread1 blocks on the 'checked' lock, this time, it completely
338
unlocks the lockdir, allowing Lock2 to acquire the lock.
341
raise tests.KnownFailure(
342
"timing dependency in lock tests (#213182)")
344
wait_to_check_lock = Lock()
345
wait_until_checked_lock = Lock()
347
wait_to_check_lock.acquire()
348
wait_until_checked_lock.acquire()
349
note('locked check and checked locks')
351
class LockDir1(LockDir):
352
"""Use the synchronization points for the first lock."""
354
def attempt_lock(self):
355
# Once we have acquired the lock, it is okay for
356
# the other lock to check it
358
return super(LockDir1, self).attempt_lock()
360
note('lock1: releasing check lock')
361
wait_to_check_lock.release()
363
class LockDir2(LockDir):
364
"""Use the synchronization points for the second lock."""
366
def attempt_lock(self):
367
note('lock2: waiting for check lock')
368
wait_to_check_lock.acquire()
369
note('lock2: acquired check lock')
371
return super(LockDir2, self).attempt_lock()
373
note('lock2: releasing checked lock')
374
wait_until_checked_lock.release()
376
t = self.get_transport()
377
lf1 = LockDir1(t, 'test_lock')
380
lf2 = LockDir2(t, 'test_lock')
381
self.setup_log_reporter(lf2)
383
def wait_and_switch():
385
# Block until lock2 has had a chance to check
386
note('lock1: waiting 1 for checked lock')
387
wait_until_checked_lock.acquire()
388
note('lock1: acquired for checked lock')
389
note('lock1: released lockdir')
391
note('lock1: acquiring lockdir')
392
# Create a new nonce, so the lock looks different.
393
lf1.nonce = osutils.rand_chars(20)
395
note('lock1: acquired lockdir')
397
# Block until lock2 has peeked again
398
note('lock1: waiting 2 for checked lock')
399
wait_until_checked_lock.acquire()
400
note('lock1: acquired for checked lock')
401
# Now unlock, and let lock 2 grab the lock
403
wait_to_check_lock.release()
405
unlocker = Thread(target=wait_and_switch)
408
# Wait and play against the other thread
409
lf2.wait_lock(timeout=20.0, poll=0.01)
414
# There should be 2 reports, because the lock changed
415
lock_base = lf2.transport.abspath(lf2.path)
416
self.assertEqual(2, len(self._logged_reports))
417
lock_url = lf2.transport.abspath(lf2.path)
418
self.assertEqual('%s %s\n'
420
'Will continue to try until %s, unless '
421
'you press Ctrl-C.\n'
422
'See "bzr help break-lock" for more.',
423
self._logged_reports[0][0])
424
args = self._logged_reports[0][1]
425
self.assertEqual('Unable to obtain', args[0])
426
self.assertEqual('lock %s' % (lock_base,), args[1])
427
self.assertStartsWith(args[2], 'held by ')
428
self.assertStartsWith(args[3], 'locked ')
429
self.assertEndsWith(args[3], ' ago')
430
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
432
self.assertEqual('%s %s\n'
434
'Will continue to try until %s, unless '
435
'you press Ctrl-C.\n'
436
'See "bzr help break-lock" for more.',
437
self._logged_reports[1][0])
438
args = self._logged_reports[1][1]
439
self.assertEqual('Lock owner changed for', args[0])
440
self.assertEqual('lock %s' % (lock_base,), args[1])
441
self.assertStartsWith(args[2], 'held by ')
442
self.assertStartsWith(args[3], 'locked ')
443
self.assertEndsWith(args[3], ' ago')
444
self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
446
def test_40_confirm_easy(self):
447
"""Confirm a lock that's already held"""
448
t = self.get_transport()
449
lf1 = LockDir(t, 'test_lock')
452
self.addCleanup(lf1.unlock)
455
def test_41_confirm_not_held(self):
456
"""Confirm a lock that's already held"""
457
t = self.get_transport()
458
lf1 = LockDir(t, 'test_lock')
460
self.assertRaises(LockNotHeld, lf1.confirm)
462
def test_42_confirm_broken_manually(self):
463
"""Confirm a lock broken by hand"""
464
t = self.get_transport()
465
lf1 = LockDir(t, 'test_lock')
468
t.move('test_lock', 'lock_gone_now')
469
self.assertRaises(LockBroken, lf1.confirm)
471
t.move('lock_gone_now', 'test_lock')
474
def test_43_break(self):
475
"""Break a lock whose caller has forgotten it"""
476
t = self.get_transport()
477
lf1 = LockDir(t, 'test_lock')
480
# we incorrectly discard the lock object without unlocking it
482
# someone else sees it's still locked
483
lf2 = LockDir(t, 'test_lock')
484
holder_info = lf2.peek()
485
self.assertTrue(holder_info)
486
lf2.force_break(holder_info)
487
# now we should be able to take it
489
self.addCleanup(lf2.unlock)
492
def test_44_break_already_released(self):
493
"""Lock break races with regular release"""
494
t = self.get_transport()
495
lf1 = LockDir(t, 'test_lock')
498
# someone else sees it's still locked
499
lf2 = LockDir(t, 'test_lock')
500
holder_info = lf2.peek()
501
# in the interim the lock is released
503
# break should succeed
504
lf2.force_break(holder_info)
505
# now we should be able to take it
507
self.addCleanup(lf2.unlock)
510
def test_45_break_mismatch(self):
511
"""Lock break races with someone else acquiring it"""
512
t = self.get_transport()
513
lf1 = LockDir(t, 'test_lock')
516
# someone else sees it's still locked
517
lf2 = LockDir(t, 'test_lock')
518
holder_info = lf2.peek()
519
# in the interim the lock is released
521
lf3 = LockDir(t, 'test_lock')
523
# break should now *fail*
524
self.assertRaises(LockBreakMismatch, lf2.force_break,
528
def test_46_fake_read_lock(self):
529
t = self.get_transport()
530
lf1 = LockDir(t, 'test_lock')
535
def test_50_lockdir_representation(self):
536
"""Check the on-disk representation of LockDirs is as expected.
538
There should always be a top-level directory named by the lock.
539
When the lock is held, there should be a lockname/held directory
540
containing an info file.
542
t = self.get_transport()
543
lf1 = LockDir(t, 'test_lock')
545
self.assertTrue(t.has('test_lock'))
547
self.assertTrue(t.has('test_lock/held/info'))
549
self.assertFalse(t.has('test_lock/held/info'))
551
def test_break_lock(self):
552
# the ui based break_lock routine should Just Work (tm)
553
ld1 = self.get_lock()
554
ld2 = self.get_lock()
557
# do this without IO redirection to ensure it doesn't prompt.
558
self.assertRaises(AssertionError, ld1.break_lock)
559
orig_factory = bzrlib.ui.ui_factory
560
bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
563
self.assertRaises(LockBroken, ld1.unlock)
565
bzrlib.ui.ui_factory = orig_factory
567
def test_break_lock_corrupt_info(self):
568
"""break_lock works even if the info file is corrupt (and tells the UI
572
ld2 = self.get_lock()
575
ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
576
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
579
def get_boolean(self, prompt):
580
self.prompts.append(('boolean', prompt))
582
ui = LoggingUIFactory()
583
orig_factory = bzrlib.ui.ui_factory
584
bzrlib.ui.ui_factory = ui
587
self.assertLength(1, ui.prompts)
588
self.assertEqual('boolean', ui.prompts[0][0])
589
self.assertStartsWith(ui.prompts[0][1], 'Break (corrupt LockDir')
590
self.assertRaises(LockBroken, ld.unlock)
592
bzrlib.ui.ui_factory = orig_factory
594
def test_break_lock_missing_info(self):
595
"""break_lock works even if the info file is missing (and tells the UI
599
ld2 = self.get_lock()
602
ld.transport.delete('test_lock/held/info')
603
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
606
def get_boolean(self, prompt):
607
self.prompts.append(('boolean', prompt))
609
ui = LoggingUIFactory()
610
orig_factory = bzrlib.ui.ui_factory
611
bzrlib.ui.ui_factory = ui
614
self.assertRaises(LockBroken, ld.unlock)
615
self.assertLength(0, ui.prompts)
617
bzrlib.ui.ui_factory = orig_factory
618
# Suppress warnings due to ld not being unlocked
619
# XXX: if lock_broken hook was invoked in this case, this hack would
620
# not be necessary. - Andrew Bennetts, 2010-09-06.
621
del self._lock_actions[:]
623
def test_create_missing_base_directory(self):
624
"""If LockDir.path doesn't exist, it can be created
626
Some people manually remove the entire lock/ directory trying
627
to unlock a stuck repository/branch/etc. Rather than failing
628
after that, just create the lock directory when needed.
630
t = self.get_transport()
631
lf1 = LockDir(t, 'test_lock')
634
self.failUnless(t.has('test_lock'))
637
self.failIf(t.has('test_lock'))
639
# This will create 'test_lock' if it needs to
641
self.failUnless(t.has('test_lock'))
642
self.failUnless(t.has('test_lock/held/info'))
645
self.failIf(t.has('test_lock/held/info'))
647
def test__format_lock_info(self):
648
ld1 = self.get_lock()
652
info_list = ld1._format_lock_info(ld1.peek())
655
self.assertEqual(info_list[0], u'jrandom@example.com')
656
# info_list[1] is hostname. we skip this.
657
self.assertContainsRe(info_list[2], '^\d+$') # pid
658
self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
660
def test_lock_without_email(self):
661
global_config = config.GlobalConfig()
662
# Intentionally has no email address
663
global_config.set_user_option('email', 'User Identity')
664
ld1 = self.get_lock()
669
def test_lock_permission(self):
670
self.requireFeature(features.not_running_as_root)
671
if not osutils.supports_posix_readonly():
672
raise tests.TestSkipped('Cannot induce a permission failure')
673
ld1 = self.get_lock()
674
lock_path = ld1.transport.local_abspath('test_lock')
676
osutils.make_readonly(lock_path)
677
self.assertRaises(errors.LockFailed, ld1.attempt_lock)
679
def test_lock_by_token(self):
680
ld1 = self.get_lock()
681
token = ld1.lock_write()
682
self.addCleanup(ld1.unlock)
683
self.assertNotEqual(None, token)
684
ld2 = self.get_lock()
685
t2 = ld2.lock_write(token)
686
self.addCleanup(ld2.unlock)
687
self.assertEqual(token, t2)
689
def test_lock_with_buggy_rename(self):
690
# test that lock acquisition handles servers which pretend they
691
# renamed correctly but that actually fail
692
t = transport.get_transport('brokenrename+' + self.get_url())
693
ld1 = LockDir(t, 'test_lock')
696
ld2 = LockDir(t, 'test_lock')
697
# we should fail to lock
698
e = self.assertRaises(errors.LockContention, ld2.attempt_lock)
699
# now the original caller should succeed in unlocking
701
# and there should be nothing left over
702
self.assertEquals([], t.list_dir('test_lock'))
704
def test_failed_lock_leaves_no_trash(self):
705
# if we fail to acquire the lock, we don't leave pending directories
706
# behind -- https://bugs.launchpad.net/bzr/+bug/109169
707
ld1 = self.get_lock()
708
ld2 = self.get_lock()
709
# should be nothing before we start
711
t = self.get_transport().clone('test_lock')
713
self.assertEquals(a, t.list_dir('.'))
715
# when held, that's all we see
717
self.addCleanup(ld1.unlock)
719
# second guy should fail
720
self.assertRaises(errors.LockContention, ld2.attempt_lock)
724
def test_no_lockdir_info(self):
725
"""We can cope with empty info files."""
726
# This seems like a fairly common failure case - see
727
# <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
728
# Processes are often interrupted after opening the file
729
# before the actual contents are committed.
730
t = self.get_transport()
732
t.mkdir('test_lock/held')
733
t.put_bytes('test_lock/held/info', '')
734
lf = LockDir(t, 'test_lock')
736
formatted_info = lf._format_lock_info(info)
738
['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
741
def test_corrupt_lockdir_info(self):
742
"""We can cope with corrupt (and thus unparseable) info files."""
743
# This seems like a fairly common failure case too - see
744
# <https://bugs.edge.launchpad.net/bzr/+bug/619872> for instance.
745
# In particular some systems tend to fill recently created files with
746
# nul bytes after recovering from a system crash.
747
t = self.get_transport()
749
t.mkdir('test_lock/held')
750
t.put_bytes('test_lock/held/info', '\0')
751
lf = LockDir(t, 'test_lock')
752
self.assertRaises(errors.LockCorrupt, lf.peek)
753
# Currently attempt_lock gives LockContention, but LockCorrupt would be
754
# a reasonable result too.
756
(errors.LockCorrupt, errors.LockContention), lf.attempt_lock)
757
self.assertRaises(errors.LockCorrupt, lf.validate_token, 'fake token')
759
def test_missing_lockdir_info(self):
760
"""We can cope with absent info files."""
761
t = self.get_transport()
763
t.mkdir('test_lock/held')
764
lf = LockDir(t, 'test_lock')
765
# In this case we expect the 'not held' result from peek, because peek
766
# cannot be expected to notice that there is a 'held' directory with no
768
self.assertEqual(None, lf.peek())
769
# And lock/unlock may work or give LockContention (but not any other
773
except LockContention:
774
# LockContention is ok, and expected on Windows
777
# no error is ok, and expected on POSIX (because POSIX allows
778
# os.rename over an empty directory).
780
# Currently raises TokenMismatch, but LockCorrupt would be reasonable
783
(errors.TokenMismatch, errors.LockCorrupt),
784
lf.validate_token, 'fake token')
787
class TestLockDirHooks(TestCaseWithTransport):
790
super(TestLockDirHooks, self).setUp()
794
return LockDir(self.get_transport(), 'test_lock')
796
def record_hook(self, result):
797
self._calls.append(result)
799
def test_LockDir_acquired_success(self):
800
# the LockDir.lock_acquired hook fires when a lock is acquired.
801
LockDir.hooks.install_named_hook('lock_acquired',
802
self.record_hook, 'record_hook')
805
self.assertEqual([], self._calls)
806
result = ld.attempt_lock()
807
lock_path = ld.transport.abspath(ld.path)
808
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
810
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
812
def test_LockDir_acquired_fail(self):
813
# the LockDir.lock_acquired hook does not fire on failure.
816
ld2 = self.get_lock()
818
# install a lock hook now, when the disk lock is locked
819
LockDir.hooks.install_named_hook('lock_acquired',
820
self.record_hook, 'record_hook')
821
self.assertRaises(errors.LockContention, ld.attempt_lock)
822
self.assertEqual([], self._calls)
824
self.assertEqual([], self._calls)
826
def test_LockDir_released_success(self):
827
# the LockDir.lock_released hook fires when a lock is acquired.
828
LockDir.hooks.install_named_hook('lock_released',
829
self.record_hook, 'record_hook')
832
self.assertEqual([], self._calls)
833
result = ld.attempt_lock()
834
self.assertEqual([], self._calls)
836
lock_path = ld.transport.abspath(ld.path)
837
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
839
def test_LockDir_released_fail(self):
840
# the LockDir.lock_released hook does not fire on failure.
843
ld2 = self.get_lock()
845
ld2.force_break(ld2.peek())
846
LockDir.hooks.install_named_hook('lock_released',
847
self.record_hook, 'record_hook')
848
self.assertRaises(LockBroken, ld.unlock)
849
self.assertEqual([], self._calls)
851
def test_LockDir_broken_success(self):
852
# the LockDir.lock_broken hook fires when a lock is broken.
855
ld2 = self.get_lock()
856
result = ld.attempt_lock()
857
LockDir.hooks.install_named_hook('lock_broken',
858
self.record_hook, 'record_hook')
859
ld2.force_break(ld2.peek())
860
lock_path = ld.transport.abspath(ld.path)
861
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
863
def test_LockDir_broken_failure(self):
864
# the LockDir.lock_broken hook does not fires when a lock is already
868
ld2 = self.get_lock()
869
result = ld.attempt_lock()
870
holder_info = ld2.peek()
872
LockDir.hooks.install_named_hook('lock_broken',
873
self.record_hook, 'record_hook')
874
ld2.force_break(holder_info)
875
lock_path = ld.transport.abspath(ld.path)
876
self.assertEqual([], self._calls)
added
removed
bzrlib/tests/test_lockdir.py
