← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

 * ``LockDir`` lock acquisition and release now trigger hooks allowing
   introspection of locking activity. (Robert Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
25
25
from bzrlib import (
26
26
    config,
27
27
    errors,
 
28
    lock,
28
29
    osutils,
29
30
    tests,
30
31
    transport,
645
646
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
646
647
        # no kibble
647
648
        check_dir(['held'])
 
649
 
 
650
    def record_hook(self, result):
 
651
        self._calls.append(result)
 
652
 
 
653
    def reset_hooks(self):
 
654
        self._old_hooks = lock.hooks
 
655
        self.addCleanup(self.restore_hooks)
 
656
        lock.hooks = lock.PhysicalLockHooks()
 
657
 
 
658
    def restore_hooks(self):
 
659
        lock.hooks = self._old_hooks
 
660
 
 
661
    def test_PhysicalLock_dot_acquired_success(self):
 
662
        # the PhysicalLock.acquired hook fires when a lock is acquired.
 
663
        self._calls = []
 
664
        self.reset_hooks()
 
665
        lock.hooks.install_hook('acquired', self.record_hook)
 
666
        ld = self.get_lock()
 
667
        ld.create()
 
668
        self.assertEqual([], self._calls)
 
669
        result = ld.attempt_lock()
 
670
        self.assertEqual([lock.LockResult(ld, result)], self._calls)
 
671
        ld.unlock()
 
672
        self.assertEqual([lock.LockResult(ld, result)], self._calls)
 
673
 
 
674
    def test_PhysicalLock_dot_acquired_fail(self):
 
675
        # the PhysicalLock.acquired hook does not fire on failure.
 
676
        self._calls = []
 
677
        self.reset_hooks()
 
678
        ld = self.get_lock()
 
679
        ld.create()
 
680
        ld2 = self.get_lock()
 
681
        ld2.attempt_lock()
 
682
        # install a lock hook now, when the disk lock is locked
 
683
        lock.hooks.install_hook('acquired', self.record_hook)
 
684
        self.assertRaises(errors.LockContention, ld.attempt_lock)
 
685
        self.assertEqual([], self._calls)
 
686
        ld2.unlock()
 
687
        self.assertEqual([], self._calls)
 
688
 
 
689
    def test_PhysicalLock_dot_released_success(self):
 
690
        # the PhysicalLock.released hook fires when a lock is acquired.
 
691
        self._calls = []
 
692
        self.reset_hooks()
 
693
        lock.hooks.install_hook('released', self.record_hook)
 
694
        ld = self.get_lock()
 
695
        ld.create()
 
696
        self.assertEqual([], self._calls)
 
697
        result = ld.attempt_lock()
 
698
        self.assertEqual([], self._calls)
 
699
        ld.unlock()
 
700
        self.assertEqual([lock.LockResult(ld, result)], self._calls)
 
701
 
 
702
    def test_PhysicalLock_dot_released_fail(self):
 
703
        # the PhysicalLock.released hook does not fire on failure.
 
704
        self._calls = []
 
705
        self.reset_hooks()
 
706
        ld = self.get_lock()
 
707
        ld.create()
 
708
        ld2 = self.get_lock()
 
709
        ld.attempt_lock()
 
710
        ld2.force_break(ld2.peek())
 
711
        lock.hooks.install_hook('released', self.record_hook)
 
712
        self.assertRaises(LockBroken, ld.unlock)
 
713
        self.assertEqual([], self._calls)