~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

Introduce a new lock_broken hook.

* tests/test_lockdir.py:
(TestLockDirHooks): Separate hook related tests from other lock
tests.
(TestLockDirHooks.test_LockDir_broken_success,
TestLockDirHooks.test_LockDir_broken_failure): Add tests for the
new lock_broken hook.

* lockdir.py:
(LockDir.force_break): Fire lock_broken hook.

* lock.py:
(LockHooks.__init__): Add lock_broken hook.

Show diffs side-by-side

added added

removed removed

Lines of Context:
670
670
        # no kibble
671
671
        check_dir(['held'])
672
672
 
 
673
 
 
674
class TestLockDirHooks(TestCaseWithTransport):
 
675
 
 
676
    def setUp(self):
 
677
        super(TestLockDirHooks, self).setUp()
 
678
        self._calls = []
 
679
        self._old_hooks = lock.Lock.hooks
 
680
 
 
681
        def restore():
 
682
            lock.Lock.hooks = self._old_hooks
 
683
        self.addCleanup(restore)
 
684
        lock.Lock.hooks = lock.LockHooks()
 
685
 
 
686
    def get_lock(self):
 
687
        return LockDir(self.get_transport(), 'test_lock')
 
688
 
673
689
    def record_hook(self, result):
674
690
        self._calls.append(result)
675
691
 
676
 
    def reset_hooks(self):
677
 
        self._old_hooks = lock.Lock.hooks
678
 
        self.addCleanup(self.restore_hooks)
679
 
        lock.Lock.hooks = lock.LockHooks()
680
 
 
681
 
    def restore_hooks(self):
682
 
        lock.Lock.hooks = self._old_hooks
683
 
 
684
692
    def test_LockDir_acquired_success(self):
685
693
        # the LockDir.lock_acquired hook fires when a lock is acquired.
686
 
        self._calls = []
687
 
        self.reset_hooks()
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()
691
697
        ld.create()
692
698
        self.assertEqual([], self._calls)
698
704
 
699
705
    def test_LockDir_acquired_fail(self):
700
706
        # the LockDir.lock_acquired hook does not fire on failure.
701
 
        self._calls = []
702
 
        self.reset_hooks()
703
707
        ld = self.get_lock()
704
708
        ld.create()
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)
712
716
        ld2.unlock()
714
718
 
715
719
    def test_LockDir_released_success(self):
716
720
        # the LockDir.lock_released hook fires when a lock is acquired.
717
 
        self._calls = []
718
 
        self.reset_hooks()
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()
722
724
        ld.create()
723
725
        self.assertEqual([], self._calls)
729
731
 
730
732
    def test_LockDir_released_fail(self):
731
733
        # the LockDir.lock_released hook does not fire on failure.
732
 
        self._calls = []
733
 
        self.reset_hooks()
734
734
        ld = self.get_lock()
735
735
        ld.create()
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)
 
743
 
 
744
    def test_LockDir_broken_success(self):
 
745
        # the LockDir.lock_broken hook fires when a lock is broken.
 
746
        ld = self.get_lock()
 
747
        ld.create()
 
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)
 
755
 
 
756
    def test_LockDir_broken_failure(self):
 
757
        # the LockDir.lock_broken hook does not fires when a lock is already
 
758
        # released.
 
759
        ld = self.get_lock()
 
760
        ld.create()
 
761
        ld2 = self.get_lock()
 
762
        result = ld.attempt_lock()
 
763
        holder_info = ld2.peek()
 
764
        ld.unlock()
 
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)