/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-01-12 18:39:25 UTC
  • mfrom: (5598.1.1 bzr)
  • Revision ID: pqm@pqm.ubuntu.com-20110112183925-rnn12p6xp91441r0
(vila) Fix PEP-8 violation in PythonVersionInfoBuilder's output (Soren
 Hansen)

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