← 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: Vincent Ladeuil
  • Date: 2010-02-09 17:27:46 UTC
  • mto: (5029.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5030.
  • Revision ID: v.ladeuil+lp@free.fr-20100209172746-6f4mvlnr2mac807j
Move NoSmartTransportServer to bzrlib.tests.test_server

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, 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
39
39
    LockNotHeld,
40
40
    )
41
41
from bzrlib.lockdir import LockDir
42
 
from bzrlib.tests import (features, TestCaseWithTransport)
 
42
from bzrlib.tests import TestCaseWithTransport
43
43
from bzrlib.trace import note
44
44
 
45
45
# These tests sometimes use threads to test the behaviour of lock files with
191
191
                    "took %f seconds to detect lock contention" % (after - before))
192
192
        finally:
193
193
            lf1.unlock()
 
194
        lock_base = lf2.transport.abspath(lf2.path)
194
195
        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}')
 
196
        lock_url = lf2.transport.abspath(lf2.path)
 
197
        self.assertEqual('%s %s\n'
 
198
                         '%s\n%s\n'
 
199
                         'Will continue to try until %s, unless '
 
200
                         'you press Ctrl-C.\n'
 
201
                         'See "bzr help break-lock" for more.',
 
202
                         self._logged_reports[0][0])
 
203
        args = self._logged_reports[0][1]
 
204
        self.assertEqual('Unable to obtain', args[0])
 
205
        self.assertEqual('lock %s' % (lock_base,), args[1])
 
206
        self.assertStartsWith(args[2], 'held by ')
 
207
        self.assertStartsWith(args[3], 'locked ')
 
208
        self.assertEndsWith(args[3], ' ago')
 
209
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
209
210
 
210
211
    def test_31_lock_wait_easy(self):
211
212
        """Succeed when waiting on a lock with no contention.
564
565
        finally:
565
566
            bzrlib.ui.ui_factory = orig_factory
566
567
 
567
 
    def test_break_lock_corrupt_info(self):
568
 
        """break_lock works even if the info file is corrupt (and tells the UI
569
 
        that it is corrupt).
570
 
        """
571
 
        ld = self.get_lock()
572
 
        ld2 = self.get_lock()
573
 
        ld.create()
574
 
        ld.lock_write()
575
 
        ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
576
 
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
577
 
            def __init__(self):
578
 
                self.prompts = []
579
 
            def get_boolean(self, prompt):
580
 
                self.prompts.append(('boolean', prompt))
581
 
                return True
582
 
        ui = LoggingUIFactory()
583
 
        orig_factory = bzrlib.ui.ui_factory
584
 
        bzrlib.ui.ui_factory = ui
585
 
        try:
586
 
            ld2.break_lock()
587
 
            self.assertLength(1, ui.prompts)
588
 
            self.assertEqual('boolean', ui.prompts[0][0])
589
 
            self.assertStartsWith(ui.prompts[0][1], 'Break (corrupt LockDir')
590
 
            self.assertRaises(LockBroken, ld.unlock)
591
 
        finally:
592
 
            bzrlib.ui.ui_factory = orig_factory
593
 
 
594
 
    def test_break_lock_missing_info(self):
595
 
        """break_lock works even if the info file is missing (and tells the UI
596
 
        that it is corrupt).
597
 
        """
598
 
        ld = self.get_lock()
599
 
        ld2 = self.get_lock()
600
 
        ld.create()
601
 
        ld.lock_write()
602
 
        ld.transport.delete('test_lock/held/info')
603
 
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
604
 
            def __init__(self):
605
 
                self.prompts = []
606
 
            def get_boolean(self, prompt):
607
 
                self.prompts.append(('boolean', prompt))
608
 
                return True
609
 
        ui = LoggingUIFactory()
610
 
        orig_factory = bzrlib.ui.ui_factory
611
 
        bzrlib.ui.ui_factory = ui
612
 
        try:
613
 
            ld2.break_lock()
614
 
            self.assertRaises(LockBroken, ld.unlock)
615
 
            self.assertLength(0, ui.prompts)
616
 
        finally:
617
 
            bzrlib.ui.ui_factory = orig_factory
618
 
        # Suppress warnings due to ld not being unlocked
619
 
        # XXX: if lock_broken hook was invoked in this case, this hack would
620
 
        # not be necessary.  - Andrew Bennetts, 2010-09-06.
621
 
        del self._lock_actions[:]
622
 
 
623
568
    def test_create_missing_base_directory(self):
624
569
        """If LockDir.path doesn't exist, it can be created
625
570
 
652
597
            info_list = ld1._format_lock_info(ld1.peek())
653
598
        finally:
654
599
            ld1.unlock()
655
 
        self.assertEqual(info_list[0], u'jrandom@example.com')
656
 
        # info_list[1] is hostname. we skip this.
657
 
        self.assertContainsRe(info_list[2], '^\d+$') # pid
658
 
        self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
 
600
        self.assertEqual('lock %s' % (ld1.transport.abspath(ld1.path),),
 
601
                         info_list[0])
 
602
        self.assertContainsRe(info_list[1],
 
603
                              r'^held by .* on host .* \[process #\d*\]$')
 
604
        self.assertContainsRe(info_list[2], r'locked \d+ seconds? ago$')
659
605
 
660
606
    def test_lock_without_email(self):
661
607
        global_config = config.GlobalConfig()
667
613
        ld1.unlock()
668
614
 
669
615
    def test_lock_permission(self):
670
 
        self.requireFeature(features.not_running_as_root)
671
616
        if not osutils.supports_posix_readonly():
672
617
            raise tests.TestSkipped('Cannot induce a permission failure')
673
618
        ld1 = self.get_lock()
721
666
        # no kibble
722
667
        check_dir(['held'])
723
668
 
724
 
    def test_no_lockdir_info(self):
725
 
        """We can cope with empty info files."""
726
 
        # This seems like a fairly common failure case - see
727
 
        # <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
728
 
        # Processes are often interrupted after opening the file
729
 
        # before the actual contents are committed.
730
 
        t = self.get_transport()
731
 
        t.mkdir('test_lock')
732
 
        t.mkdir('test_lock/held')
733
 
        t.put_bytes('test_lock/held/info', '')
734
 
        lf = LockDir(t, 'test_lock')
735
 
        info = lf.peek()
736
 
        formatted_info = lf._format_lock_info(info)
737
 
        self.assertEquals(
738
 
            ['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
739
 
            formatted_info)
740
 
 
741
 
    def test_corrupt_lockdir_info(self):
742
 
        """We can cope with corrupt (and thus unparseable) info files."""
743
 
        # This seems like a fairly common failure case too - see
744
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/619872> for instance.
745
 
        # In particular some systems tend to fill recently created files with
746
 
        # nul bytes after recovering from a system crash.
747
 
        t = self.get_transport()
748
 
        t.mkdir('test_lock')
749
 
        t.mkdir('test_lock/held')
750
 
        t.put_bytes('test_lock/held/info', '\0')
751
 
        lf = LockDir(t, 'test_lock')
752
 
        self.assertRaises(errors.LockCorrupt, lf.peek)
753
 
        # Currently attempt_lock gives LockContention, but LockCorrupt would be
754
 
        # a reasonable result too.
755
 
        self.assertRaises(
756
 
            (errors.LockCorrupt, errors.LockContention), lf.attempt_lock)
757
 
        self.assertRaises(errors.LockCorrupt, lf.validate_token, 'fake token')
758
 
 
759
 
    def test_missing_lockdir_info(self):
760
 
        """We can cope with absent info files."""
761
 
        t = self.get_transport()
762
 
        t.mkdir('test_lock')
763
 
        t.mkdir('test_lock/held')
764
 
        lf = LockDir(t, 'test_lock')
765
 
        # In this case we expect the 'not held' result from peek, because peek
766
 
        # cannot be expected to notice that there is a 'held' directory with no
767
 
        # 'info' file.
768
 
        self.assertEqual(None, lf.peek())
769
 
        # And lock/unlock may work or give LockContention (but not any other
770
 
        # error).
771
 
        try:
772
 
            lf.attempt_lock()
773
 
        except LockContention:
774
 
            # LockContention is ok, and expected on Windows
775
 
            pass
776
 
        else:
777
 
            # no error is ok, and expected on POSIX (because POSIX allows
778
 
            # os.rename over an empty directory).
779
 
            lf.unlock()
780
 
        # Currently raises TokenMismatch, but LockCorrupt would be reasonable
781
 
        # too.
782
 
        self.assertRaises(
783
 
            (errors.TokenMismatch, errors.LockCorrupt),
784
 
            lf.validate_token, 'fake token')
785
 
 
786
669
 
787
670
class TestLockDirHooks(TestCaseWithTransport):
788
671