191
184
# it should only take about 0.4 seconds, but we allow more time in
192
185
# case the machine is heavily loaded
193
186
self.assertTrue(after - before <= 8.0,
194
"took %f seconds to detect lock contention" % (after - before))
187
"took %f seconds to detect lock contention" % (after - before))
197
190
self.assertEqual(1, len(self._logged_reports))
198
self.assertContainsRe(self._logged_reports[0][0],
199
r'Unable to obtain lock .* held by jrandom@example\.com on .*'
200
r' \(process #\d+\), acquired .* ago\.\n'
201
r'Will continue to try until \d{2}:\d{2}:\d{2}, unless '
202
r'you press Ctrl-C.\n'
203
r'See "bzr help break-lock" for more.')
191
self.assertEqual(self._logged_reports[0][0],
192
'%s lock %s held by %s\n'
193
'at %s [process #%s], acquired %s.\n'
194
'Will continue to try until %s, unless '
195
'you press Ctrl-C.\n'
196
'See "bzr help break-lock" for more.')
197
start, lock_url, user, hostname, pid, time_ago, deadline_str = \
198
self._logged_reports[0][1]
199
self.assertEqual(start, u'Unable to obtain')
200
self.assertEqual(user, u'jrandom@example.com')
202
self.assertContainsRe(pid, r'\d+')
203
self.assertContainsRe(time_ago, r'.* ago')
204
self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
205
206
def test_31_lock_wait_easy(self):
206
207
"""Succeed when waiting on a lock with no contention.
422
417
self.assertFalse(t.has('test_lock/held/info'))
424
def test_display_form(self):
419
def test__format_lock_info(self):
425
420
ld1 = self.get_lock()
429
info_list = ld1.peek().to_readable_dict()
424
info_list = ld1._format_lock_info(ld1.peek())
432
self.assertEqual(info_list['user'], u'jrandom@example.com')
433
self.assertContainsRe(info_list['pid'], '^\d+$')
434
self.assertContainsRe(info_list['time_ago'], r'^\d+ seconds? ago$')
427
self.assertEqual(info_list[0], u'jrandom@example.com')
428
# info_list[1] is hostname. we skip this.
429
self.assertContainsRe(info_list[2], '^\d+$') # pid
430
self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
436
432
def test_lock_without_email(self):
437
433
global_config = config.GlobalConfig()
653
646
ld2.force_break(holder_info)
654
647
lock_path = ld.transport.abspath(ld.path)
655
648
self.assertEqual([], self._calls)
658
class TestLockHeldInfo(TestCase):
659
"""Can get information about the lock holder, and detect whether they're
663
info = LockHeldInfo.for_this_process(None)
664
self.assertContainsRe(repr(info), r"LockHeldInfo\(.*\)")
666
def test_unicode(self):
667
info = LockHeldInfo.for_this_process(None)
668
self.assertContainsRe(unicode(info),
669
r'held by .* on .* \(process #\d+\), acquired .* ago')
671
def test_is_locked_by_this_process(self):
672
info = LockHeldInfo.for_this_process(None)
673
self.assertTrue(info.is_locked_by_this_process())
675
def test_is_not_locked_by_this_process(self):
676
info = LockHeldInfo.for_this_process(None)
677
info.info_dict['pid'] = '123123123123123'
678
self.assertFalse(info.is_locked_by_this_process())
680
def test_lock_holder_live_process(self):
681
"""Detect that the holder (this process) is still running."""
682
info = LockHeldInfo.for_this_process(None)
683
self.assertFalse(info.is_lock_holder_known_dead())
685
def test_lock_holder_dead_process(self):
686
"""Detect that the holder (this process) is still running."""
687
info = LockHeldInfo.for_this_process(None)
688
info.info_dict['pid'] = '123123123'
689
if sys.platform == 'win32':
691
'live lock holder detection not implemented yet on win32')
692
self.assertTrue(info.is_lock_holder_known_dead())
694
def test_lock_holder_other_machine(self):
695
"""The lock holder isn't here so we don't know if they're alive."""
696
info = LockHeldInfo.for_this_process(None)
697
info.info_dict['hostname'] = 'egg.example.com'
698
info.info_dict['pid'] = '123123123'
699
self.assertFalse(info.is_lock_holder_known_dead())
701
def test_lock_holder_other_user(self):
702
"""Only auto-break locks held by this user."""
703
info = LockHeldInfo.for_this_process(None)
704
info.info_dict['user'] = 'notme@example.com'
705
info.info_dict['pid'] = '123123123'
706
self.assertFalse(info.is_lock_holder_known_dead())
708
def test_no_good_hostname(self):
709
"""Correctly handle ambiguous hostnames.
711
If the lock's recorded with just 'localhost' we can't really trust
712
it's the same 'localhost'. (There are quite a few of them. :-)
713
So even if the process is known not to be alive, we can't say that's
716
self.overrideAttr(lockdir, 'get_host_name',
718
info = LockHeldInfo.for_this_process(None)
719
info.info_dict['pid'] = '123123123'
720
self.assertFalse(info.is_lock_holder_known_dead())
723
class TestStaleLockDir(TestCaseWithTransport):
724
"""Can automatically break stale locks.
726
:see: https://bugs.launchpad.net/bzr/+bug/220464
729
def test_auto_break_stale_lock(self):
730
"""Locks safely known to be stale are just cleaned up.
732
This generates a warning but no other user interaction.
734
# This is off by default at present; see the discussion in the bug.
735
# If you change the default, don't forget to update the docs.
736
config.GlobalConfig().set_user_option('locks.steal_dead', True)
737
# Create a lock pretending to come from a different nonexistent
738
# process on the same machine.
739
l1 = LockDir(self.get_transport(), 'a',
740
extra_holder_info={'pid': '12312313'})
741
token_1 = l1.attempt_lock()
742
l2 = LockDir(self.get_transport(), 'a')
743
token_2 = l2.attempt_lock()
744
# l1 will notice its lock was stolen.
745
self.assertRaises(errors.LockBroken,
749
def test_auto_break_stale_lock_configured_off(self):
750
"""Automatic breaking can be turned off"""
751
l1 = LockDir(self.get_transport(), 'a',
752
extra_holder_info={'pid': '12312313'})
753
token_1 = l1.attempt_lock()
754
self.addCleanup(l1.unlock)
755
l2 = LockDir(self.get_transport(), 'a')
756
# This fails now, because dead lock breaking is off by default.
757
self.assertRaises(LockContention,
759
# and it's in fact not broken
added
removed
bzrlib/tests/test_lockdir.py
