← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_branch.py

Abbreviate pack_stat struct format to '>6L'

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_branch.py
16
16
 
17
17
"""Tests for the Branch facility that are not interface  tests.
18
18
 
19
 
For interface tests see tests/per_branch/*.py.
 
19
For interface tests see `tests/per_branch/*.py`.
20
20
 
21
21
For concrete class tests see this file, and for meta-branch tests
22
22
also see this file.
112
112
        """See BzrBranchFormat.get_format_string()."""
113
113
        return "Sample branch format."
114
114
 
115
 
    def initialize(self, a_bzrdir, name=None, repository=None):
 
115
    def initialize(self, a_bzrdir, name=None, repository=None,
 
116
                   append_revisions_only=None):
116
117
        """Format 4 branches cannot be created."""
117
118
        t = a_bzrdir.get_branch_transport(self, name=name)
118
119
        t.put_bytes('format', self.get_format_string())
137
138
        """See BzrBranchFormat.get_format_string()."""
138
139
        return SampleSupportedBranchFormatString
139
140
 
140
 
    def initialize(self, a_bzrdir, name=None):
 
141
    def initialize(self, a_bzrdir, name=None, append_revisions_only=None):
141
142
        t = a_bzrdir.get_branch_transport(self, name=name)
142
143
        t.put_bytes('format', self.get_format_string())
143
144
        return 'A branch'
652
653
        if value is not None:
653
654
            self.config.set_user_option('append_revisions_only', value)
654
655
        self.assertEqual(expected_value,
655
 
                         self.branch._get_append_revisions_only())
 
656
                         self.branch.get_append_revisions_only())
656
657
 
657
658
    def test_valid_append_revisions_only(self):
658
659
        self.assertEquals(None,
709
710
        r.new_revid = "same-revid"
710
711
        f = StringIO()
711
712
        r.report(f)
712
 
        self.assertEqual("No revisions to pull.\n", f.getvalue())
713
 
 
714
 
 
715
 
class _StubLockable(object):
716
 
    """Helper for TestRunWithWriteLockedTarget."""
717
 
 
718
 
    def __init__(self, calls, unlock_exc=None):
719
 
        self.calls = calls
720
 
        self.unlock_exc = unlock_exc
721
 
 
722
 
    def lock_write(self):
723
 
        self.calls.append('lock_write')
724
 
 
725
 
    def unlock(self):
726
 
        self.calls.append('unlock')
727
 
        if self.unlock_exc is not None:
728
 
            raise self.unlock_exc
729
 
 
730
 
 
731
 
class _ErrorFromCallable(Exception):
732
 
    """Helper for TestRunWithWriteLockedTarget."""
733
 
 
734
 
 
735
 
class _ErrorFromUnlock(Exception):
736
 
    """Helper for TestRunWithWriteLockedTarget."""
737
 
 
738
 
 
739
 
class TestRunWithWriteLockedTarget(tests.TestCase):
740
 
    """Tests for _run_with_write_locked_target."""
741
 
 
742
 
    def setUp(self):
743
 
        tests.TestCase.setUp(self)
744
 
        self._calls = []
745
 
 
746
 
    def func_that_returns_ok(self):
747
 
        self._calls.append('func called')
748
 
        return 'ok'
749
 
 
750
 
    def func_that_raises(self):
751
 
        self._calls.append('func called')
752
 
        raise _ErrorFromCallable()
753
 
 
754
 
    def test_success_unlocks(self):
755
 
        lockable = _StubLockable(self._calls)
756
 
        result = _mod_branch._run_with_write_locked_target(
757
 
            lockable, self.func_that_returns_ok)
758
 
        self.assertEqual('ok', result)
759
 
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
760
 
 
761
 
    def test_exception_unlocks_and_propagates(self):
762
 
        lockable = _StubLockable(self._calls)
763
 
        self.assertRaises(_ErrorFromCallable,
764
 
                          _mod_branch._run_with_write_locked_target,
765
 
                          lockable, self.func_that_raises)
766
 
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
767
 
 
768
 
    def test_callable_succeeds_but_error_during_unlock(self):
769
 
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
770
 
        self.assertRaises(_ErrorFromUnlock,
771
 
                          _mod_branch._run_with_write_locked_target,
772
 
                          lockable, self.func_that_returns_ok)
773
 
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
774
 
 
775
 
    def test_error_during_unlock_does_not_mask_original_error(self):
776
 
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
777
 
        self.assertRaises(_ErrorFromCallable,
778
 
                          _mod_branch._run_with_write_locked_target,
779
 
                          lockable, self.func_that_raises)
780
 
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
713
        self.assertEqual("No revisions or tags to pull.\n", f.getvalue())
 
714