189
184
# it should only take about 0.4 seconds, but we allow more time in
190
185
# case the machine is heavily loaded
191
186
self.assertTrue(after - before <= 8.0,
192
"took %f seconds to detect lock contention" % (after - before))
187
"took %f seconds to detect lock contention" % (after - before))
195
190
self.assertEqual(1, len(self._logged_reports))
196
self.assertContainsRe(self._logged_reports[0][0],
197
r'Unable to obtain lock .* held by jrandom@example\.com on .*'
198
r' \(process #\d+\), acquired .* ago\.\n'
199
r'Will continue to try until \d{2}:\d{2}:\d{2}, unless '
200
r'you press Ctrl-C.\n'
201
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}')
203
206
def test_31_lock_wait_easy(self):
204
207
"""Succeed when waiting on a lock with no contention.
420
417
self.assertFalse(t.has('test_lock/held/info'))
422
def test_display_form(self):
419
def test__format_lock_info(self):
423
420
ld1 = self.get_lock()
427
info_list = ld1.peek().to_readable_dict()
424
info_list = ld1._format_lock_info(ld1.peek())
430
self.assertEqual(info_list['user'], u'jrandom@example.com')
431
self.assertContainsRe(info_list['pid'], '^\d+$')
432
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
434
432
def test_lock_without_email(self):
435
433
global_config = config.GlobalConfig()
652
646
ld2.force_break(holder_info)
653
647
lock_path = ld.transport.abspath(ld.path)
654
648
self.assertEqual([], self._calls)
657
class TestLockHeldInfo(TestCase):
658
"""Can get information about the lock holder, and detect whether they're
662
info = LockHeldInfo.for_this_process(None)
663
self.assertContainsRe(repr(info), r"LockHeldInfo\(.*\)")
665
def test_unicode(self):
666
info = LockHeldInfo.for_this_process(None)
667
self.assertContainsRe(unicode(info),
668
r'held by .* on .* \(process #\d+\), acquired .* ago')
670
def test_is_locked_by_this_process(self):
671
info = LockHeldInfo.for_this_process(None)
672
self.assertTrue(info.is_locked_by_this_process())
674
def test_is_not_locked_by_this_process(self):
675
info = LockHeldInfo.for_this_process(None)
676
info.info_dict['pid'] = '123123123123123'
677
self.assertFalse(info.is_locked_by_this_process())
679
def test_lock_holder_live_process(self):
680
"""Detect that the holder (this process) is still running."""
681
info = LockHeldInfo.for_this_process(None)
682
self.assertFalse(info.is_lock_holder_known_dead())
684
def test_lock_holder_dead_process(self):
685
"""Detect that the holder (this process) is still running."""
686
self.overrideAttr(lockdir, 'get_host_name',
687
lambda: 'aproperhostname')
688
info = LockHeldInfo.for_this_process(None)
689
info.info_dict['pid'] = '123123123'
690
self.assertTrue(info.is_lock_holder_known_dead())
692
def test_lock_holder_other_machine(self):
693
"""The lock holder isn't here so we don't know if they're alive."""
694
info = LockHeldInfo.for_this_process(None)
695
info.info_dict['hostname'] = 'egg.example.com'
696
info.info_dict['pid'] = '123123123'
697
self.assertFalse(info.is_lock_holder_known_dead())
699
def test_lock_holder_other_user(self):
700
"""Only auto-break locks held by this user."""
701
info = LockHeldInfo.for_this_process(None)
702
info.info_dict['user'] = 'notme@example.com'
703
info.info_dict['pid'] = '123123123'
704
self.assertFalse(info.is_lock_holder_known_dead())
706
def test_no_good_hostname(self):
707
"""Correctly handle ambiguous hostnames.
709
If the lock's recorded with just 'localhost' we can't really trust
710
it's the same 'localhost'. (There are quite a few of them. :-)
711
So even if the process is known not to be alive, we can't say that's
714
self.overrideAttr(lockdir, 'get_host_name',
716
info = LockHeldInfo.for_this_process(None)
717
info.info_dict['pid'] = '123123123'
718
self.assertFalse(info.is_lock_holder_known_dead())
721
class TestStaleLockDir(TestCaseWithTransport):
722
"""Can automatically break stale locks.
724
:see: https://bugs.launchpad.net/bzr/+bug/220464
727
def test_auto_break_stale_lock(self):
728
"""Locks safely known to be stale are just cleaned up.
730
This generates a warning but no other user interaction.
732
self.overrideAttr(lockdir, 'get_host_name',
733
lambda: 'aproperhostname')
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
