« back to all changes in this revision
Viewing changes to bzrlib/tests/branch_implementations/test_locking.py
- browse files at revision 1711.8.5
- compare with another revision
- compare with revision 2255.7.57
- download diff
- download tarball
- view history from revision 1711.8.5
- Committer: John Arbash Meinel
- Date: 2006-07-07 15:20:33 UTC
- mto: This revision was merged to the branch mainline in revision 1846.
- Revision ID: john@arbash-meinel.com-20060707152033-b9840586a3e0ab49
Move the new locking tests into their own files, and move the helper functions into a test helper.
- files added:
- bzrlib/tests/LockHelpers.py
- files modified:
- bzrlib/errors.py
added
removed
bzrlib/tests/branch_implementations/test_locking.py
1
# Copyright (C) 2006 Canonical Ltd
2
#
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
7
#
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
12
#
13
# You should have received a copy of the GNU General Public License
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
16
17
"""Test locks across all branch implemenations"""
18
19
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
20
from bzrlib.tests.LockHelpers import TestPreventLocking, LockWrapper
21
22
23
class TestBranchLocking(TestCaseWithBranch):
24
25
def get_instrumented_branch(self):
26
"""Get a Branch object which has been instrumented"""
27
# TODO: jam 20060630 It may be that not all formats have a
28
# 'control_files' member. So we should fail gracefully if
29
# not there. But assuming it has them lets us test the exact
30
# lock/unlock order.
31
self.locks = []
32
b = LockWrapper(self.locks, self.get_branch(), 'b')
33
b.repository = LockWrapper(self.locks, b.repository, 'r')
34
bcf = b.control_files
35
rcf = b.repository.control_files
36
37
# Look out for branch types that reuse their control files
38
self.combined_control = bcf is rcf
39
40
b.control_files = LockWrapper(self.locks, b.control_files, 'bc')
41
b.repository.control_files = \
42
LockWrapper(self.locks, b.repository.control_files, 'rc')
43
return b
44
45
def test_01_lock_read(self):
46
# Test that locking occurs in the correct order
47
b = self.get_instrumented_branch()
48
49
self.assertFalse(b.is_locked())
50
self.assertFalse(b.repository.is_locked())
51
b.lock_read()
52
try:
53
self.assertTrue(b.is_locked())
54
self.assertTrue(b.repository.is_locked())
55
finally:
56
b.unlock()
57
self.assertFalse(b.is_locked())
58
self.assertFalse(b.repository.is_locked())
59
60
self.assertEqual([('b', 'lr', True),
61
('r', 'lr', True),
62
('rc', 'lr', True),
63
('bc', 'lr', True),
64
('b', 'ul', True),
65
('bc', 'ul', True),
66
('r', 'ul', True),
67
('rc', 'ul', True),
68
], self.locks)
69
70
def test_02_lock_write(self):
71
# Test that locking occurs in the correct order
72
b = self.get_instrumented_branch()
73
74
self.assertFalse(b.is_locked())
75
self.assertFalse(b.repository.is_locked())
76
b.lock_write()
77
try:
78
self.assertTrue(b.is_locked())
79
self.assertTrue(b.repository.is_locked())
80
finally:
81
b.unlock()
82
self.assertFalse(b.is_locked())
83
self.assertFalse(b.repository.is_locked())
84
85
self.assertEqual([('b', 'lw', True),
86
('r', 'lw', True),
87
('rc', 'lw', True),
88
('bc', 'lw', True),
89
('b', 'ul', True),
90
('bc', 'ul', True),
91
('r', 'ul', True),
92
('rc', 'ul', True),
93
], self.locks)
94
95
def test_03_lock_fail_unlock_repo(self):
96
# Make sure branch.unlock() is called, even if there is a
97
# failure while unlocking the repository.
98
b = self.get_instrumented_branch()
99
b.repository.disable_unlock()
100
101
self.assertFalse(b.is_locked())
102
self.assertFalse(b.repository.is_locked())
103
b.lock_write()
104
try:
105
self.assertTrue(b.is_locked())
106
self.assertTrue(b.repository.is_locked())
107
self.assertRaises(TestPreventLocking, b.unlock)
108
if self.combined_control:
109
self.assertTrue(b.is_locked())
110
else:
111
self.assertFalse(b.is_locked())
112
self.assertTrue(b.repository.is_locked())
113
114
# We unlock the branch control files, even if
115
# we fail to unlock the repository
116
self.assertEqual([('b', 'lw', True),
117
('r', 'lw', True),
118
('rc', 'lw', True),
119
('bc', 'lw', True),
120
('b', 'ul', True),
121
('bc', 'ul', True),
122
('r', 'ul', False),
123
], self.locks)
124
125
finally:
126
# For cleanup purposes, make sure we are unlocked
127
b.repository._other.unlock()
128
129
def test_04_lock_fail_unlock_control(self):
130
# Make sure repository.unlock() is called, if we fail to unlock self
131
b = self.get_instrumented_branch()
132
b.control_files.disable_unlock()
133
134
self.assertFalse(b.is_locked())
135
self.assertFalse(b.repository.is_locked())
136
b.lock_write()
137
try:
138
self.assertTrue(b.is_locked())
139
self.assertTrue(b.repository.is_locked())
140
self.assertRaises(TestPreventLocking, b.unlock)
141
self.assertTrue(b.is_locked())
142
if self.combined_control:
143
self.assertTrue(b.repository.is_locked())
144
else:
145
self.assertFalse(b.repository.is_locked())
146
147
# We unlock the repository even if
148
# we fail to unlock the control files
149
self.assertEqual([('b', 'lw', True),
150
('r', 'lw', True),
151
('rc', 'lw', True),
152
('bc', 'lw', True),
153
('b', 'ul', True),
154
('bc', 'ul', False),
155
('r', 'ul', True),
156
('rc', 'ul', True),
157
], self.locks)
158
159
finally:
160
# For cleanup purposes, make sure we are unlocked
161
b.control_files._other.unlock()
162
163
def test_05_lock_read_fail_repo(self):
164
# Test that the branch is not locked if it cannot lock the repository
165
b = self.get_instrumented_branch()
166
b.repository.disable_lock_read()
167
168
self.assertRaises(TestPreventLocking, b.lock_read)
169
self.assertFalse(b.is_locked())
170
self.assertFalse(b.repository.is_locked())
171
172
self.assertEqual([('b', 'lr', True),
173
('r', 'lr', False),
174
], self.locks)
175
176
def test_06_lock_write_fail_repo(self):
177
# Test that the branch is not locked if it cannot lock the repository
178
b = self.get_instrumented_branch()
179
b.repository.disable_lock_write()
180
181
self.assertRaises(TestPreventLocking, b.lock_write)
182
self.assertFalse(b.is_locked())
183
self.assertFalse(b.repository.is_locked())
184
185
self.assertEqual([('b', 'lw', True),
186
('r', 'lw', False),
187
], self.locks)
188
189
def test_07_lock_read_fail_control(self):
190
# Test the repository is unlocked if we can't lock self
191
b = self.get_instrumented_branch()
192
b.control_files.disable_lock_read()
193
194
self.assertRaises(TestPreventLocking, b.lock_read)
195
self.assertFalse(b.is_locked())
196
self.assertFalse(b.repository.is_locked())
197
198
self.assertEqual([('b', 'lr', True),
199
('r', 'lr', True),
200
('rc', 'lr', True),
201
('bc', 'lr', False),
202
('r', 'ul', True),
203
('rc', 'ul', True),
204
], self.locks)
205
206
def test_08_lock_write_fail_control(self):
207
# Test the repository is unlocked if we can't lock self
208
b = self.get_instrumented_branch()
209
b.control_files.disable_lock_write()
210
211
self.assertRaises(TestPreventLocking, b.lock_write)
212
self.assertFalse(b.is_locked())
213
self.assertFalse(b.repository.is_locked())
214
215
self.assertEqual([('b', 'lw', True),
216
('r', 'lw', True),
217
('rc', 'lw', True),
218
('bc', 'lw', False),
219
('r', 'ul', True),
220
('rc', 'ul', True),
221
], self.locks)
222
Loggerhead is a web-based interface for Breezy
Version: 2.0.1