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
45
45
# These tests sometimes use threads to test the behaviour of lock files with
565
565
bzrlib.ui.ui_factory = orig_factory
567
def test_break_lock_corrupt_info(self):
568
"""break_lock works even if the info file is corrupt (and tells the UI
572
ld2 = self.get_lock()
575
ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
576
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
579
def get_boolean(self, prompt):
580
self.prompts.append(('boolean', prompt))
582
ui = LoggingUIFactory()
583
orig_factory = bzrlib.ui.ui_factory
584
bzrlib.ui.ui_factory = ui
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)
592
bzrlib.ui.ui_factory = orig_factory
594
def test_break_lock_missing_info(self):
595
"""break_lock works even if the info file is missing (and tells the UI
599
ld2 = self.get_lock()
602
ld.transport.delete('test_lock/held/info')
603
class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
606
def get_boolean(self, prompt):
607
self.prompts.append(('boolean', prompt))
609
ui = LoggingUIFactory()
610
orig_factory = bzrlib.ui.ui_factory
611
bzrlib.ui.ui_factory = ui
614
self.assertRaises(LockBroken, ld.unlock)
615
self.assertLength(0, ui.prompts)
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[:]
623
567
def test_create_missing_base_directory(self):
624
568
"""If LockDir.path doesn't exist, it can be created
738
681
['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
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()
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.
756
(errors.LockCorrupt, errors.LockContention), lf.attempt_lock)
757
self.assertRaises(errors.LockCorrupt, lf.validate_token, 'fake token')
759
def test_missing_lockdir_info(self):
760
"""We can cope with absent info files."""
761
t = self.get_transport()
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
768
self.assertEqual(None, lf.peek())
769
# And lock/unlock may work or give LockContention (but not any other
773
except LockContention:
774
# LockContention is ok, and expected on Windows
777
# no error is ok, and expected on POSIX (because POSIX allows
778
# os.rename over an empty directory).
780
# Currently raises TokenMismatch, but LockCorrupt would be reasonable
783
(errors.TokenMismatch, errors.LockCorrupt),
784
lf.validate_token, 'fake token')
787
685
class TestLockDirHooks(TestCaseWithTransport):