← 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: 2009-04-09 23:12:55 UTC
  • mfrom: (3920.2.37 dpush)
  • Revision ID: pqm@pqm.ubuntu.com-20090409231255-o8w1g2q3igiyf8b2
(Jelmer) Add the dpush command.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006, 2007, 2008 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)
146
145
        # lock is held, should get some info on it
147
146
        info1 = lf1.peek()
148
147
        self.assertEqual(set(info1.keys()),
162
161
        lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
163
162
        self.assertEqual(lf2.peek(), None)
164
163
        lf1.attempt_lock()
165
 
        self.addCleanup(lf1.unlock)
166
164
        info2 = lf2.peek()
167
165
        self.assertTrue(info2)
168
166
        self.assertEqual(info2['nonce'], lf1.nonce)
191
189
                    "took %f seconds to detect lock contention" % (after - before))
192
190
        finally:
193
191
            lf1.unlock()
 
192
        lock_base = lf2.transport.abspath(lf2.path)
194
193
        self.assertEqual(1, len(self._logged_reports))
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}')
 
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')
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
 
                         'See "bzr help break-lock" for more.',
 
421
                         'you press Ctrl-C\n'
 
422
                         'If you\'re sure that it\'s not being '
 
423
                         'modified, use bzr break-lock %s',
423
424
                         self._logged_reports[0][0])
424
425
        args = self._logged_reports[0][1]
425
426
        self.assertEqual('Unable to obtain', args[0])
432
433
        self.assertEqual('%s %s\n'
433
434
                         '%s\n%s\n'
434
435
                         'Will continue to try until %s, unless '
435
 
                         'you press Ctrl-C.\n'
436
 
                         'See "bzr help break-lock" for more.',
 
436
                         'you press Ctrl-C\n'
 
437
                         'If you\'re sure that it\'s not being '
 
438
                         'modified, use bzr break-lock %s',
437
439
                         self._logged_reports[1][0])
438
440
        args = self._logged_reports[1][1]
439
441
        self.assertEqual('Lock owner changed for', args[0])
449
451
        lf1 = LockDir(t, 'test_lock')
450
452
        lf1.create()
451
453
        lf1.attempt_lock()
452
 
        self.addCleanup(lf1.unlock)
453
454
        lf1.confirm()
454
455
 
455
456
    def test_41_confirm_not_held(self):
467
468
        lf1.attempt_lock()
468
469
        t.move('test_lock', 'lock_gone_now')
469
470
        self.assertRaises(LockBroken, lf1.confirm)
470
 
        # Clean up
471
 
        t.move('lock_gone_now', 'test_lock')
472
 
        lf1.unlock()
473
471
 
474
472
    def test_43_break(self):
475
473
        """Break a lock whose caller has forgotten it"""
486
484
        lf2.force_break(holder_info)
487
485
        # now we should be able to take it
488
486
        lf2.attempt_lock()
489
 
        self.addCleanup(lf2.unlock)
490
487
        lf2.confirm()
491
488
 
492
489
    def test_44_break_already_released(self):
504
501
        lf2.force_break(holder_info)
505
502
        # now we should be able to take it
506
503
        lf2.attempt_lock()
507
 
        self.addCleanup(lf2.unlock)
508
504
        lf2.confirm()
509
505
 
510
506
    def test_45_break_mismatch(self):
557
553
        # do this without IO redirection to ensure it doesn't prompt.
558
554
        self.assertRaises(AssertionError, ld1.break_lock)
559
555
        orig_factory = bzrlib.ui.ui_factory
560
 
        bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
 
556
        # silent ui - no need for stdout
 
557
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
558
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
561
559
        try:
562
560
            ld2.break_lock()
563
561
            self.assertRaises(LockBroken, ld1.unlock)
596
594
            info_list = ld1._format_lock_info(ld1.peek())
597
595
        finally:
598
596
            ld1.unlock()
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
 
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$')
603
602
 
604
603
    def test_lock_without_email(self):
605
604
        global_config = config.GlobalConfig()
622
621
    def test_lock_by_token(self):
623
622
        ld1 = self.get_lock()
624
623
        token = ld1.lock_write()
625
 
        self.addCleanup(ld1.unlock)
626
624
        self.assertNotEqual(None, token)
627
625
        ld2 = self.get_lock()
628
626
        t2 = ld2.lock_write(token)
629
 
        self.addCleanup(ld2.unlock)
630
627
        self.assertEqual(token, t2)
631
628
 
632
629
    def test_lock_with_buggy_rename(self):
657
654
        check_dir([])
658
655
        # when held, that's all we see
659
656
        ld1.attempt_lock()
660
 
        self.addCleanup(ld1.unlock)
661
657
        check_dir(['held'])
662
658
        # second guy should fail
663
659
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
664
660
        # no kibble
665
661
        check_dir(['held'])
666
662
 
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
 
 
694
663
    def record_hook(self, result):
695
664
        self._calls.append(result)
696
665
 
 
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
 
697
674
    def test_LockDir_acquired_success(self):
698
675
        # the LockDir.lock_acquired hook fires when a lock is acquired.
 
676
        self._calls = []
 
677
        self.reset_hooks()
699
678
        LockDir.hooks.install_named_hook('lock_acquired',
700
 
                                         self.record_hook, 'record_hook')
 
679
            self.record_hook, 'record_hook')
701
680
        ld = self.get_lock()
702
681
        ld.create()
703
682
        self.assertEqual([], self._calls)
709
688
 
710
689
    def test_LockDir_acquired_fail(self):
711
690
        # the LockDir.lock_acquired hook does not fire on failure.
 
691
        self._calls = []
 
692
        self.reset_hooks()
712
693
        ld = self.get_lock()
713
694
        ld.create()
714
695
        ld2 = self.get_lock()
715
696
        ld2.attempt_lock()
716
697
        # install a lock hook now, when the disk lock is locked
717
698
        LockDir.hooks.install_named_hook('lock_acquired',
718
 
                                         self.record_hook, 'record_hook')
 
699
            self.record_hook, 'record_hook')
719
700
        self.assertRaises(errors.LockContention, ld.attempt_lock)
720
701
        self.assertEqual([], self._calls)
721
702
        ld2.unlock()
723
704
 
724
705
    def test_LockDir_released_success(self):
725
706
        # the LockDir.lock_released hook fires when a lock is acquired.
 
707
        self._calls = []
 
708
        self.reset_hooks()
726
709
        LockDir.hooks.install_named_hook('lock_released',
727
 
                                         self.record_hook, 'record_hook')
 
710
            self.record_hook, 'record_hook')
728
711
        ld = self.get_lock()
729
712
        ld.create()
730
713
        self.assertEqual([], self._calls)
736
719
 
737
720
    def test_LockDir_released_fail(self):
738
721
        # the LockDir.lock_released hook does not fire on failure.
 
722
        self._calls = []
 
723
        self.reset_hooks()
739
724
        ld = self.get_lock()
740
725
        ld.create()
741
726
        ld2 = self.get_lock()
742
727
        ld.attempt_lock()
743
728
        ld2.force_break(ld2.peek())
744
729
        LockDir.hooks.install_named_hook('lock_released',
745
 
                                         self.record_hook, 'record_hook')
 
730
            self.record_hook, 'record_hook')
746
731
        self.assertRaises(LockBroken, ld.unlock)
747
732
        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)