/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/per_branch/test_locking.py

  • Committer: Andrew Bennetts
  • Date: 2010-01-12 03:53:21 UTC
  • mfrom: (4948 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4964.
  • Revision ID: andrew.bennetts@canonical.com-20100112035321-hofpz5p10224ryj3
Merge lp:bzr, resolving conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_branch/test_locking.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Test locks across all branch implemenations"""
18
18
 
20
20
from bzrlib.branch import BzrBranchFormat4
21
21
from bzrlib.bzrdir import RemoteBzrDirFormat
22
22
from bzrlib.tests import TestSkipped
23
 
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
24
23
from bzrlib.tests.lock_helpers import TestPreventLocking, LockWrapper
 
24
from bzrlib.tests.per_branch.test_branch import TestCaseWithBranch
25
25
 
26
26
 
27
27
class TestBranchLocking(TestCaseWithBranch):
32
32
 
33
33
    def get_instrumented_branch(self):
34
34
        """Get a Branch object which has been instrumented"""
35
 
        # TODO: jam 20060630 It may be that not all formats have a 
 
35
        # TODO: jam 20060630 It may be that not all formats have a
36
36
        # 'control_files' member. So we should fail gracefully if
37
 
        # not there. But assuming it has them lets us test the exact 
 
37
        # not there. But assuming it has them lets us test the exact
38
38
        # lock/unlock order.
39
39
        self.locks = []
40
40
        b = LockWrapper(self.locks, self.get_branch(), 'b')
54
54
        if self.combined_control:
55
55
            # instrument the repository control files too to ensure its worked
56
56
            # with correctly. When they are not shared, we trust the repository
57
 
            # API and only instrument the repository itself. 
 
57
            # API and only instrument the repository itself.
58
58
            b.repository.control_files = \
59
59
                LockWrapper(self.locks, b.repository.control_files, 'rc')
60
60
        return b
139
139
        try:
140
140
            self.assertTrue(b.is_locked())
141
141
            self.assertTrue(b.repository.is_locked())
142
 
            self.assertRaises(TestPreventLocking, b.unlock)
 
142
            self.assertLogsError(TestPreventLocking, b.unlock)
143
143
            if self.combined_control:
144
144
                self.assertTrue(b.is_locked())
145
145
            else:
146
146
                self.assertFalse(b.is_locked())
147
147
            self.assertTrue(b.repository.is_locked())
148
148
 
149
 
            # We unlock the branch control files, even if 
 
149
            # We unlock the branch control files, even if
150
150
            # we fail to unlock the repository
151
151
            if self.combined_control:
152
152
                self.assertEqual([('b', 'lw', True),
171
171
            b.repository._other.unlock()
172
172
 
173
173
    def test_04_lock_fail_unlock_control(self):
174
 
        # Make sure repository.unlock() is called, if we fail to unlock self
 
174
        # Make sure repository.unlock() is not called, if we fail to unlock
 
175
        # self leaving ourselves still locked, so that attempts to recover
 
176
        # don't encounter an unlocked repository.
175
177
        b = self.get_instrumented_branch()
176
178
        b.control_files.disable_unlock()
177
179
 
181
183
        try:
182
184
            self.assertTrue(b.is_locked())
183
185
            self.assertTrue(b.repository.is_locked())
184
 
            self.assertRaises(TestPreventLocking, b.unlock)
 
186
            self.assertLogsError(TestPreventLocking, b.unlock)
185
187
            self.assertTrue(b.is_locked())
186
 
            if self.combined_control:
187
 
                self.assertTrue(b.repository.is_locked())
188
 
            else:
189
 
                self.assertFalse(b.repository.is_locked())
 
188
            self.assertTrue(b.repository.is_locked())
190
189
 
191
 
            # We unlock the repository even if 
 
190
            # We unlock the repository even if
192
191
            # we fail to unlock the control files
193
192
            if self.combined_control:
194
193
                self.assertEqual([('b', 'lw', True),
206
205
                                  ('bc', 'lw', True),
207
206
                                  ('b', 'ul', True),
208
207
                                  ('bc', 'ul', False),
209
 
                                  ('r', 'ul', True),
210
208
                                 ], self.locks)
211
209
 
212
210
        finally:
223
221
        self.assertFalse(b.repository.is_locked())
224
222
 
225
223
        self.assertEqual([('b', 'lr', True),
226
 
                          ('r', 'lr', False), 
 
224
                          ('r', 'lr', False),
227
225
                         ], self.locks)
228
226
 
229
227
    def test_06_lock_write_fail_repo(self):
236
234
        self.assertFalse(b.repository.is_locked())
237
235
 
238
236
        self.assertEqual([('b', 'lw', True),
239
 
                          ('r', 'lw', False), 
 
237
                          ('r', 'lw', False),
240
238
                         ], self.locks)
241
239
 
242
240
    def test_07_lock_read_fail_control(self):
434
432
            branch.unlock()
435
433
        # We should be unable to relock the repo.
436
434
        self.assertRaises(errors.LockContention, branch.lock_write)
 
435
        # Cleanup
 
436
        branch.lock_write(token)
 
437
        branch.dont_leave_lock_in_place()
 
438
        branch.unlock()
437
439
 
438
440
    def test_dont_leave_lock_in_place(self):
439
441
        branch = self.make_branch('b')