46
43
# so won't modernize them now. - mbp 20080430
47
44
class _TestLockableFiles_mixin(object):
46
def test_read_write(self):
47
self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
48
self.assertRaises(NoSuchFile,
50
deprecated_in((1, 5, 0)),
51
self.lockable.get_utf8, 'foo')
52
self.lockable.lock_write()
54
unicode_string = u'bar\u1234'
55
self.assertEqual(4, len(unicode_string))
56
byte_string = unicode_string.encode('utf-8')
57
self.assertEqual(6, len(byte_string))
58
self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo',
59
StringIO(unicode_string))
60
self.lockable.put('foo', StringIO(byte_string))
61
self.assertEqual(byte_string,
62
self.lockable.get('foo').read())
63
unicode_stream = self.applyDeprecated(
64
deprecated_in((1, 5, 0)),
65
self.lockable.get_utf8,
67
self.assertEqual(unicode_string,
68
unicode_stream.read())
69
self.assertRaises(BzrBadParameterNotString,
70
self.lockable.put_utf8,
72
StringIO(unicode_string)
74
self.lockable.put_utf8('bar', unicode_string)
75
unicode_stream = self.applyDeprecated(
76
deprecated_in((1, 5, 0)),
77
self.lockable.get_utf8,
79
self.assertEqual(unicode_string,
80
unicode_stream.read())
81
self.assertEqual(byte_string,
82
self.lockable.get('bar').read())
83
self.lockable.put_bytes('raw', 'raw\xffbytes')
84
self.assertEqual('raw\xffbytes',
85
self.lockable.get('raw').read())
87
self.lockable.unlock()
90
self.lockable.lock_read()
92
self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
93
StringIO('bar\u1234'))
95
self.lockable.unlock()
49
97
def test_transactions(self):
50
98
self.assertIs(self.lockable.get_transaction().__class__,
51
99
PassThroughTransaction)
99
148
def test_lock_write_returns_None_refuses_token(self):
100
149
token = self.lockable.lock_write()
101
self.addCleanup(self.lockable.unlock)
102
if token is not None:
103
# This test does not apply, because this lockable supports
105
raise TestNotApplicable("%r uses tokens" % (self.lockable,))
106
self.assertRaises(errors.TokenLockingNotSupported,
107
self.lockable.lock_write, token='token')
151
if token is not None:
152
# This test does not apply, because this lockable supports
155
self.assertRaises(errors.TokenLockingNotSupported,
156
self.lockable.lock_write, token='token')
158
self.lockable.unlock()
109
160
def test_lock_write_returns_token_when_given_token(self):
110
161
token = self.lockable.lock_write()
111
self.addCleanup(self.lockable.unlock)
113
# This test does not apply, because this lockable refuses
116
new_lockable = self.get_lockable()
117
token_from_new_lockable = new_lockable.lock_write(token=token)
118
self.addCleanup(new_lockable.unlock)
119
self.assertEqual(token, token_from_new_lockable)
164
# This test does not apply, because this lockable refuses
167
new_lockable = self.get_lockable()
168
token_from_new_lockable = new_lockable.lock_write(token=token)
170
self.assertEqual(token, token_from_new_lockable)
172
new_lockable.unlock()
174
self.lockable.unlock()
121
176
def test_lock_write_raises_on_token_mismatch(self):
122
177
token = self.lockable.lock_write()
123
self.addCleanup(self.lockable.unlock)
125
# This test does not apply, because this lockable refuses
128
different_token = token + 'xxx'
129
# Re-using the same lockable instance with a different token will
130
# raise TokenMismatch.
131
self.assertRaises(errors.TokenMismatch,
132
self.lockable.lock_write, token=different_token)
133
# A separate instance for the same lockable will also raise
135
# This detects the case where a caller claims to have a lock (via
136
# the token) for an external resource, but doesn't (the token is
137
# different). Clients need a separate lock object to make sure the
138
# external resource is probed, whereas the existing lock object
140
new_lockable = self.get_lockable()
141
self.assertRaises(errors.TokenMismatch,
142
new_lockable.lock_write, token=different_token)
180
# This test does not apply, because this lockable refuses
183
different_token = token + 'xxx'
184
# Re-using the same lockable instance with a different token will
185
# raise TokenMismatch.
186
self.assertRaises(errors.TokenMismatch,
187
self.lockable.lock_write, token=different_token)
188
# A seperate instance for the same lockable will also raise
190
# This detects the case where a caller claims to have a lock (via
191
# the token) for an external resource, but doesn't (the token is
192
# different). Clients need a seperate lock object to make sure the
193
# external resource is probed, whereas the existing lock object
195
new_lockable = self.get_lockable()
196
self.assertRaises(errors.TokenMismatch,
197
new_lockable.lock_write, token=different_token)
199
self.lockable.unlock()
144
201
def test_lock_write_with_matching_token(self):
145
202
# If the token matches, so no exception is raised by lock_write.
146
203
token = self.lockable.lock_write()
147
self.addCleanup(self.lockable.unlock)
149
# This test does not apply, because this lockable refuses
152
# The same instance will accept a second lock_write if the specified
154
self.lockable.lock_write(token=token)
155
self.lockable.unlock()
156
# Calling lock_write on a new instance for the same lockable will
158
new_lockable = self.get_lockable()
159
new_lockable.lock_write(token=token)
160
new_lockable.unlock()
206
# This test does not apply, because this lockable refuses
209
# The same instance will accept a second lock_write if the specified
211
self.lockable.lock_write(token=token)
212
self.lockable.unlock()
213
# Calling lock_write on a new instance for the same lockable will
215
new_lockable = self.get_lockable()
216
new_lockable.lock_write(token=token)
217
new_lockable.unlock()
219
self.lockable.unlock()
162
221
def test_unlock_after_lock_write_with_token(self):
163
222
# If lock_write did not physically acquire the lock (because it was
164
223
# passed a token), then unlock should not physically release it.
165
224
token = self.lockable.lock_write()
166
self.addCleanup(self.lockable.unlock)
168
# This test does not apply, because this lockable refuses
171
new_lockable = self.get_lockable()
172
new_lockable.lock_write(token=token)
173
new_lockable.unlock()
174
self.assertTrue(self.lockable.get_physical_lock_status())
227
# This test does not apply, because this lockable refuses
230
new_lockable = self.get_lockable()
231
new_lockable.lock_write(token=token)
232
new_lockable.unlock()
233
self.assertTrue(self.lockable.get_physical_lock_status())
235
self.lockable.unlock()
176
237
def test_lock_write_with_token_fails_when_unlocked(self):
177
238
# Lock and unlock to get a superficially valid token. This mimics a
271
327
third_lockable.unlock()
274
# This method of adapting tests to parameters is different to
275
# the TestProviderAdapters used elsewhere, but seems simpler for this
330
# This method of adapting tests to parameters is different to
331
# the TestProviderAdapters used elsewhere, but seems simpler for this
277
333
class TestLockableFiles_TransportLock(TestCaseInTempDir,
278
334
_TestLockableFiles_mixin):
281
337
TestCaseInTempDir.setUp(self)
282
t = transport.get_transport('.')
284
self.sub_transport = t.clone('.bzr')
338
transport = get_transport('.')
339
transport.mkdir('.bzr')
340
self.sub_transport = transport.clone('.bzr')
285
341
self.lockable = self.get_lockable()
286
342
self.lockable.create_lock()
288
def stop_server(self):
289
super(TestLockableFiles_TransportLock, self).stop_server()
345
super(TestLockableFiles_TransportLock, self).tearDown()
290
346
# free the subtransport so that we do not get a 5 second
291
347
# timeout due to the SFTP connection cache.
293
del self.sub_transport
294
except AttributeError:
348
del self.sub_transport
297
350
def get_lockable(self):
298
351
return LockableFiles(self.sub_transport, 'my-lock', TransportLock)
301
354
class TestLockableFiles_LockDir(TestCaseInTempDir,
302
355
_TestLockableFiles_mixin):
added
removed
bzrlib/tests/test_lockable_files.py
