671
671
check_dir(['held'])
674
class TestLockDirHooks(TestCaseWithTransport):
677
super(TestLockDirHooks, self).setUp()
679
self._old_hooks = lock.Lock.hooks
682
lock.Lock.hooks = self._old_hooks
683
self.addCleanup(restore)
684
lock.Lock.hooks = lock.LockHooks()
687
return LockDir(self.get_transport(), 'test_lock')
673
689
def record_hook(self, result):
674
690
self._calls.append(result)
676
def reset_hooks(self):
677
self._old_hooks = lock.Lock.hooks
678
self.addCleanup(self.restore_hooks)
679
lock.Lock.hooks = lock.LockHooks()
681
def restore_hooks(self):
682
lock.Lock.hooks = self._old_hooks
684
692
def test_LockDir_acquired_success(self):
685
693
# the LockDir.lock_acquired hook fires when a lock is acquired.
688
694
LockDir.hooks.install_named_hook('lock_acquired',
689
self.record_hook, 'record_hook')
695
self.record_hook, 'record_hook')
690
696
ld = self.get_lock()
692
698
self.assertEqual([], self._calls)
699
705
def test_LockDir_acquired_fail(self):
700
706
# the LockDir.lock_acquired hook does not fire on failure.
703
707
ld = self.get_lock()
705
709
ld2 = self.get_lock()
706
710
ld2.attempt_lock()
707
711
# install a lock hook now, when the disk lock is locked
708
712
LockDir.hooks.install_named_hook('lock_acquired',
709
self.record_hook, 'record_hook')
713
self.record_hook, 'record_hook')
710
714
self.assertRaises(errors.LockContention, ld.attempt_lock)
711
715
self.assertEqual([], self._calls)
715
719
def test_LockDir_released_success(self):
716
720
# the LockDir.lock_released hook fires when a lock is acquired.
719
721
LockDir.hooks.install_named_hook('lock_released',
720
self.record_hook, 'record_hook')
722
self.record_hook, 'record_hook')
721
723
ld = self.get_lock()
723
725
self.assertEqual([], self._calls)
730
732
def test_LockDir_released_fail(self):
731
733
# the LockDir.lock_released hook does not fire on failure.
734
734
ld = self.get_lock()
736
736
ld2 = self.get_lock()
737
737
ld.attempt_lock()
738
738
ld2.force_break(ld2.peek())
739
739
LockDir.hooks.install_named_hook('lock_released',
740
self.record_hook, 'record_hook')
740
self.record_hook, 'record_hook')
741
741
self.assertRaises(LockBroken, ld.unlock)
742
742
self.assertEqual([], self._calls)
744
def test_LockDir_broken_success(self):
745
# the LockDir.lock_broken hook fires when a lock is broken.
748
ld2 = self.get_lock()
749
result = ld.attempt_lock()
750
LockDir.hooks.install_named_hook('lock_broken',
751
self.record_hook, 'record_hook')
752
ld2.force_break(ld2.peek())
753
lock_path = ld.transport.abspath(ld.path)
754
self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
756
def test_LockDir_broken_failure(self):
757
# the LockDir.lock_broken hook does not fires when a lock is already
761
ld2 = self.get_lock()
762
result = ld.attempt_lock()
763
holder_info = ld2.peek()
765
LockDir.hooks.install_named_hook('lock_broken',
766
self.record_hook, 'record_hook')
767
ld2.force_break(holder_info)
768
lock_path = ld.transport.abspath(ld.path)
769
self.assertEqual([], self._calls)