← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_branch/test_locking.py

  • Committer: Robert Collins
  • Date: 2010-04-08 04:34:03 UTC
  • mfrom: (5138 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5139.
  • Revision ID: robertc@robertcollins.net-20100408043403-56z0d07vdqrx7f3t
Update bugfix for 528114 to trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_branch/test_locking.py
1
 
# Copyright (C) 2006 Canonical Ltd
 
1
# Copyright (C) 2006-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
16
16
 
17
17
"""Test locks across all branch implemenations"""
18
18
 
19
 
from bzrlib import errors
20
 
from bzrlib.branch import BzrBranchFormat4
21
 
from bzrlib.bzrdir import RemoteBzrDirFormat
22
 
from bzrlib.tests import TestSkipped
23
 
from bzrlib.tests.lock_helpers import TestPreventLocking, LockWrapper
24
 
from bzrlib.tests.per_branch.test_branch import TestCaseWithBranch
25
 
 
26
 
 
27
 
class TestBranchLocking(TestCaseWithBranch):
 
19
from bzrlib import (
 
20
    branch as _mod_branch,
 
21
    errors,
 
22
    tests,
 
23
    )
 
24
from bzrlib.tests import (
 
25
    lock_helpers,
 
26
    per_branch,
 
27
    )
 
28
 
 
29
 
 
30
class TestBranchLocking(per_branch.TestCaseWithBranch):
28
31
 
29
32
    def setUp(self):
30
 
        TestCaseWithBranch.setUp(self)
 
33
        super(TestBranchLocking, self).setUp()
31
34
        self.reduceLockdirTimeout()
32
35
 
33
36
    def get_instrumented_branch(self):
37
40
        # not there. But assuming it has them lets us test the exact
38
41
        # lock/unlock order.
39
42
        self.locks = []
40
 
        b = LockWrapper(self.locks, self.get_branch(), 'b')
41
 
        b.repository = LockWrapper(self.locks, b.repository, 'r')
 
43
        b = lock_helpers.LockWrapper(self.locks, self.get_branch(), 'b')
 
44
        b.repository = lock_helpers.LockWrapper(self.locks, b.repository, 'r')
42
45
        bcf = b.control_files
43
46
        rcf = getattr(b.repository, 'control_files', None)
44
47
        if rcf is None:
47
50
            # Look out for branch types that reuse their control files
48
51
            self.combined_control = bcf is rcf
49
52
        try:
50
 
            b.control_files = LockWrapper(self.locks, b.control_files, 'bc')
 
53
            b.control_files = lock_helpers.LockWrapper(
 
54
                self.locks, b.control_files, 'bc')
51
55
        except AttributeError:
52
56
            # RemoteBranch seems to trigger this.
53
 
            raise TestSkipped("Could not instrument branch control files.")
 
57
            raise tests.TestSkipped(
 
58
                'Could not instrument branch control files.')
54
59
        if self.combined_control:
55
60
            # instrument the repository control files too to ensure its worked
56
61
            # with correctly. When they are not shared, we trust the repository
57
62
            # API and only instrument the repository itself.
58
 
            b.repository.control_files = \
59
 
                LockWrapper(self.locks, b.repository.control_files, 'rc')
 
63
            b.repository.control_files = lock_helpers.LockWrapper(
 
64
                self.locks, b.repository.control_files, 'rc')
60
65
        return b
61
66
 
62
67
    def test_01_lock_read(self):
139
144
        try:
140
145
            self.assertTrue(b.is_locked())
141
146
            self.assertTrue(b.repository.is_locked())
142
 
            self.assertLogsError(TestPreventLocking, b.unlock)
 
147
            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)
143
148
            if self.combined_control:
144
149
                self.assertTrue(b.is_locked())
145
150
            else:
183
188
        try:
184
189
            self.assertTrue(b.is_locked())
185
190
            self.assertTrue(b.repository.is_locked())
186
 
            self.assertLogsError(TestPreventLocking, b.unlock)
 
191
            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)
187
192
            self.assertTrue(b.is_locked())
188
193
            self.assertTrue(b.repository.is_locked())
189
194
 
216
221
        b = self.get_instrumented_branch()
217
222
        b.repository.disable_lock_read()
218
223
 
219
 
        self.assertRaises(TestPreventLocking, b.lock_read)
 
224
        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)
220
225
        self.assertFalse(b.is_locked())
221
226
        self.assertFalse(b.repository.is_locked())
222
227
 
229
234
        b = self.get_instrumented_branch()
230
235
        b.repository.disable_lock_write()
231
236
 
232
 
        self.assertRaises(TestPreventLocking, b.lock_write)
 
237
        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)
233
238
        self.assertFalse(b.is_locked())
234
239
        self.assertFalse(b.repository.is_locked())
235
240
 
242
247
        b = self.get_instrumented_branch()
243
248
        b.control_files.disable_lock_read()
244
249
 
245
 
        self.assertRaises(TestPreventLocking, b.lock_read)
 
250
        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)
246
251
        self.assertFalse(b.is_locked())
247
252
        self.assertFalse(b.repository.is_locked())
248
253
 
266
271
        b = self.get_instrumented_branch()
267
272
        b.control_files.disable_lock_write()
268
273
 
269
 
        self.assertRaises(TestPreventLocking, b.lock_write)
 
274
        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)
270
275
        self.assertFalse(b.is_locked())
271
276
        self.assertFalse(b.repository.is_locked())
272
277
        if self.combined_control:
493
498
        branch.unlock()
494
499
 
495
500
    def test_lock_write_locks_repo_too(self):
496
 
        if isinstance(self.branch_format, BzrBranchFormat4):
 
501
        if isinstance(self.branch_format, _mod_branch.BzrBranchFormat4):
497
502
            # Branch format 4 is combined with the repository, so this test
498
503
            # doesn't apply.
499
504
            return