← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006, 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2006-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
142
142
        lf1 = LockDir(t, 'test_lock')
143
143
        lf1.create()
144
144
        lf1.attempt_lock()
 
145
        self.addCleanup(lf1.unlock)
145
146
        # lock is held, should get some info on it
146
147
        info1 = lf1.peek()
147
148
        self.assertEqual(set(info1.keys()),
161
162
        lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
162
163
        self.assertEqual(lf2.peek(), None)
163
164
        lf1.attempt_lock()
 
165
        self.addCleanup(lf1.unlock)
164
166
        info2 = lf2.peek()
165
167
        self.assertTrue(info2)
166
168
        self.assertEqual(info2['nonce'], lf1.nonce)
189
191
                    "took %f seconds to detect lock contention" % (after - before))
190
192
        finally:
191
193
            lf1.unlock()
192
 
        lock_base = lf2.transport.abspath(lf2.path)
193
194
        self.assertEqual(1, len(self._logged_reports))
194
 
        lock_url = lf2.transport.abspath(lf2.path)
195
 
        self.assertEqual('%s %s\n'
196
 
                         '%s\n%s\n'
197
 
                         'Will continue to try until %s, unless '
198
 
                         'you press Ctrl-C\n'
199
 
                         'If you\'re sure that it\'s not being '
200
 
                         'modified, use bzr break-lock %s',
201
 
                         self._logged_reports[0][0])
202
 
        args = self._logged_reports[0][1]
203
 
        self.assertEqual('Unable to obtain', args[0])
204
 
        self.assertEqual('lock %s' % (lock_base,), args[1])
205
 
        self.assertStartsWith(args[2], 'held by ')
206
 
        self.assertStartsWith(args[3], 'locked ')
207
 
        self.assertEndsWith(args[3], ' ago')
208
 
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
 
195
        self.assertEqual(self._logged_reports[0][0],
 
196
            '%s lock %s held by %s\n'
 
197
            'at %s [process #%s], acquired %s.\n'
 
198
            'Will continue to try until %s, unless '
 
199
            'you press Ctrl-C.\n'
 
200
            'See "bzr help break-lock" for more.')
 
201
        start, lock_url, user, hostname, pid, time_ago, deadline_str = \
 
202
            self._logged_reports[0][1]
 
203
        self.assertEqual(start, u'Unable to obtain')
 
204
        self.assertEqual(user, u'jrandom@example.com')
 
205
        # skip hostname
 
206
        self.assertContainsRe(pid, r'\d+')
 
207
        self.assertContainsRe(time_ago, r'.* ago')
 
208
        self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
209
209
 
210
210
    def test_31_lock_wait_easy(self):
211
211
        """Succeed when waiting on a lock with no contention.
418
418
        self.assertEqual('%s %s\n'
419
419
                         '%s\n%s\n'
420
420
                         'Will continue to try until %s, unless '
421
 
                         'you press Ctrl-C\n'
422
 
                         'If you\'re sure that it\'s not being '
423
 
                         'modified, use bzr break-lock %s',
 
421
                         'you press Ctrl-C.\n'
 
422
                         'See "bzr help break-lock" for more.',
424
423
                         self._logged_reports[0][0])
425
424
        args = self._logged_reports[0][1]
426
425
        self.assertEqual('Unable to obtain', args[0])
433
432
        self.assertEqual('%s %s\n'
434
433
                         '%s\n%s\n'
435
434
                         'Will continue to try until %s, unless '
436
 
                         'you press Ctrl-C\n'
437
 
                         'If you\'re sure that it\'s not being '
438
 
                         'modified, use bzr break-lock %s',
 
435
                         'you press Ctrl-C.\n'
 
436
                         'See "bzr help break-lock" for more.',
439
437
                         self._logged_reports[1][0])
440
438
        args = self._logged_reports[1][1]
441
439
        self.assertEqual('Lock owner changed for', args[0])
451
449
        lf1 = LockDir(t, 'test_lock')
452
450
        lf1.create()
453
451
        lf1.attempt_lock()
 
452
        self.addCleanup(lf1.unlock)
454
453
        lf1.confirm()
455
454
 
456
455
    def test_41_confirm_not_held(self):
468
467
        lf1.attempt_lock()
469
468
        t.move('test_lock', 'lock_gone_now')
470
469
        self.assertRaises(LockBroken, lf1.confirm)
 
470
        # Clean up
 
471
        t.move('lock_gone_now', 'test_lock')
 
472
        lf1.unlock()
471
473
 
472
474
    def test_43_break(self):
473
475
        """Break a lock whose caller has forgotten it"""
484
486
        lf2.force_break(holder_info)
485
487
        # now we should be able to take it
486
488
        lf2.attempt_lock()
 
489
        self.addCleanup(lf2.unlock)
487
490
        lf2.confirm()
488
491
 
489
492
    def test_44_break_already_released(self):
501
504
        lf2.force_break(holder_info)
502
505
        # now we should be able to take it
503
506
        lf2.attempt_lock()
 
507
        self.addCleanup(lf2.unlock)
504
508
        lf2.confirm()
505
509
 
506
510
    def test_45_break_mismatch(self):
553
557
        # do this without IO redirection to ensure it doesn't prompt.
554
558
        self.assertRaises(AssertionError, ld1.break_lock)
555
559
        orig_factory = bzrlib.ui.ui_factory
556
 
        # silent ui - no need for stdout
557
 
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
558
 
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
 
560
        bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
559
561
        try:
560
562
            ld2.break_lock()
561
563
            self.assertRaises(LockBroken, ld1.unlock)
594
596
            info_list = ld1._format_lock_info(ld1.peek())
595
597
        finally:
596
598
            ld1.unlock()
597
 
        self.assertEqual('lock %s' % (ld1.transport.abspath(ld1.path),),
598
 
                         info_list[0])
599
 
        self.assertContainsRe(info_list[1],
600
 
                              r'^held by .* on host .* \[process #\d*\]$')
601
 
        self.assertContainsRe(info_list[2], r'locked \d+ seconds? ago$')
 
599
        self.assertEqual(info_list[0], u'jrandom@example.com')
 
600
        # info_list[1] is hostname. we skip this.
 
601
        self.assertContainsRe(info_list[2], '^\d+$') # pid
 
602
        self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
602
603
 
603
604
    def test_lock_without_email(self):
604
605
        global_config = config.GlobalConfig()
621
622
    def test_lock_by_token(self):
622
623
        ld1 = self.get_lock()
623
624
        token = ld1.lock_write()
 
625
        self.addCleanup(ld1.unlock)
624
626
        self.assertNotEqual(None, token)
625
627
        ld2 = self.get_lock()
626
628
        t2 = ld2.lock_write(token)
 
629
        self.addCleanup(ld2.unlock)
627
630
        self.assertEqual(token, t2)
628
631
 
629
632
    def test_lock_with_buggy_rename(self):
654
657
        check_dir([])
655
658
        # when held, that's all we see
656
659
        ld1.attempt_lock()
 
660
        self.addCleanup(ld1.unlock)
657
661
        check_dir(['held'])
658
662
        # second guy should fail
659
663
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
660
664
        # no kibble
661
665
        check_dir(['held'])
662
666
 
 
667
    def test_no_lockdir_info(self):
 
668
        """We can cope with empty info files."""
 
669
        # This seems like a fairly common failure case - see
 
670
        # <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
 
671
        # Processes are often interrupted after opening the file
 
672
        # before the actual contents are committed.
 
673
        t = self.get_transport()
 
674
        t.mkdir('test_lock')
 
675
        t.mkdir('test_lock/held')
 
676
        t.put_bytes('test_lock/held/info', '')
 
677
        lf = LockDir(t, 'test_lock')
 
678
        info = lf.peek()
 
679
        formatted_info = lf._format_lock_info(info)
 
680
        self.assertEquals(
 
681
            ['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
 
682
            formatted_info)
 
683
 
 
684
 
 
685
class TestLockDirHooks(TestCaseWithTransport):
 
686
 
 
687
    def setUp(self):
 
688
        super(TestLockDirHooks, self).setUp()
 
689
        self._calls = []
 
690
 
 
691
    def get_lock(self):
 
692
        return LockDir(self.get_transport(), 'test_lock')
 
693
 
663
694
    def record_hook(self, result):
664
695
        self._calls.append(result)
665
696
 
666
 
    def reset_hooks(self):
667
 
        self._old_hooks = lock.Lock.hooks
668
 
        self.addCleanup(self.restore_hooks)
669
 
        lock.Lock.hooks = lock.LockHooks()
670
 
 
671
 
    def restore_hooks(self):
672
 
        lock.Lock.hooks = self._old_hooks
673
 
 
674
697
    def test_LockDir_acquired_success(self):
675
698
        # the LockDir.lock_acquired hook fires when a lock is acquired.
676
 
        self._calls = []
677
 
        self.reset_hooks()
678
699
        LockDir.hooks.install_named_hook('lock_acquired',
679
 
            self.record_hook, 'record_hook')
 
700
                                         self.record_hook, 'record_hook')
680
701
        ld = self.get_lock()
681
702
        ld.create()
682
703
        self.assertEqual([], self._calls)
688
709
 
689
710
    def test_LockDir_acquired_fail(self):
690
711
        # the LockDir.lock_acquired hook does not fire on failure.
691
 
        self._calls = []
692
 
        self.reset_hooks()
693
712
        ld = self.get_lock()
694
713
        ld.create()
695
714
        ld2 = self.get_lock()
696
715
        ld2.attempt_lock()
697
716
        # install a lock hook now, when the disk lock is locked
698
717
        LockDir.hooks.install_named_hook('lock_acquired',
699
 
            self.record_hook, 'record_hook')
 
718
                                         self.record_hook, 'record_hook')
700
719
        self.assertRaises(errors.LockContention, ld.attempt_lock)
701
720
        self.assertEqual([], self._calls)
702
721
        ld2.unlock()
704
723
 
705
724
    def test_LockDir_released_success(self):
706
725
        # the LockDir.lock_released hook fires when a lock is acquired.
707
 
        self._calls = []
708
 
        self.reset_hooks()
709
726
        LockDir.hooks.install_named_hook('lock_released',
710
 
            self.record_hook, 'record_hook')
 
727
                                         self.record_hook, 'record_hook')
711
728
        ld = self.get_lock()
712
729
        ld.create()
713
730
        self.assertEqual([], self._calls)
719
736
 
720
737
    def test_LockDir_released_fail(self):
721
738
        # the LockDir.lock_released hook does not fire on failure.
722
 
        self._calls = []
723
 
        self.reset_hooks()
724
739
        ld = self.get_lock()
725
740
        ld.create()
726
741
        ld2 = self.get_lock()
727
742
        ld.attempt_lock()
728
743
        ld2.force_break(ld2.peek())
729
744
        LockDir.hooks.install_named_hook('lock_released',
730
 
            self.record_hook, 'record_hook')
 
745
                                         self.record_hook, 'record_hook')
731
746
        self.assertRaises(LockBroken, ld.unlock)
732
747
        self.assertEqual([], self._calls)
 
748
 
 
749
    def test_LockDir_broken_success(self):
 
750
        # the LockDir.lock_broken hook fires when a lock is broken.
 
751
        ld = self.get_lock()
 
752
        ld.create()
 
753
        ld2 = self.get_lock()
 
754
        result = ld.attempt_lock()
 
755
        LockDir.hooks.install_named_hook('lock_broken',
 
756
                                         self.record_hook, 'record_hook')
 
757
        ld2.force_break(ld2.peek())
 
758
        lock_path = ld.transport.abspath(ld.path)
 
759
        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
 
760
 
 
761
    def test_LockDir_broken_failure(self):
 
762
        # the LockDir.lock_broken hook does not fires when a lock is already
 
763
        # released.
 
764
        ld = self.get_lock()
 
765
        ld.create()
 
766
        ld2 = self.get_lock()
 
767
        result = ld.attempt_lock()
 
768
        holder_info = ld2.peek()
 
769
        ld.unlock()
 
770
        LockDir.hooks.install_named_hook('lock_broken',
 
771
                                         self.record_hook, 'record_hook')
 
772
        ld2.force_break(holder_info)
 
773
        lock_path = ld.transport.abspath(ld.path)
 
774
        self.assertEqual([], self._calls)