43
37
# these tests are applied in each parameterized suite for LockableFiles
45
# they use an old style of parameterization, but we want to remove this class
46
# so won't modernize them now. - mbp 20080430
47
38
class _TestLockableFiles_mixin(object):
40
def test_read_write(self):
41
self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
42
self.assertRaises(NoSuchFile, self.lockable.get_utf8, 'foo')
43
self.lockable.lock_write()
45
unicode_string = u'bar\u1234'
46
self.assertEqual(4, len(unicode_string))
47
byte_string = unicode_string.encode('utf-8')
48
self.assertEqual(6, len(byte_string))
49
self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo',
50
StringIO(unicode_string))
51
self.lockable.put('foo', StringIO(byte_string))
52
self.assertEqual(byte_string,
53
self.lockable.get('foo').read())
54
self.assertEqual(unicode_string,
55
self.lockable.get_utf8('foo').read())
56
self.assertRaises(BzrBadParameterNotString,
57
self.lockable.put_utf8,
59
StringIO(unicode_string)
61
self.lockable.put_utf8('bar', unicode_string)
62
self.assertEqual(unicode_string,
63
self.lockable.get_utf8('bar').read())
64
self.assertEqual(byte_string,
65
self.lockable.get('bar').read())
66
self.lockable.put_bytes('raw', 'raw\xffbytes')
67
self.assertEqual('raw\xffbytes',
68
self.lockable.get('raw').read())
70
self.lockable.unlock()
73
self.lockable.lock_read()
75
self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
76
StringIO('bar\u1234'))
78
self.lockable.unlock()
49
80
def test_transactions(self):
50
81
self.assertIs(self.lockable.get_transaction().__class__,
51
82
PassThroughTransaction)
99
131
def test_lock_write_returns_None_refuses_token(self):
100
132
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')
134
if token is not None:
135
# This test does not apply, because this lockable supports
138
self.assertRaises(errors.TokenLockingNotSupported,
139
self.lockable.lock_write, token='token')
141
self.lockable.unlock()
109
143
def test_lock_write_returns_token_when_given_token(self):
110
144
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)
147
# This test does not apply, because this lockable refuses
150
new_lockable = self.get_lockable()
151
token_from_new_lockable = new_lockable.lock_write(token=token)
153
self.assertEqual(token, token_from_new_lockable)
155
new_lockable.unlock()
157
self.lockable.unlock()
121
159
def test_lock_write_raises_on_token_mismatch(self):
122
160
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)
163
# This test does not apply, because this lockable refuses
166
different_token = token + 'xxx'
167
# Re-using the same lockable instance with a different token will
168
# raise TokenMismatch.
169
self.assertRaises(errors.TokenMismatch,
170
self.lockable.lock_write, token=different_token)
171
# A seperate instance for the same lockable will also raise
173
# This detects the case where a caller claims to have a lock (via
174
# the token) for an external resource, but doesn't (the token is
175
# different). Clients need a seperate lock object to make sure the
176
# external resource is probed, whereas the existing lock object
178
new_lockable = self.get_lockable()
179
self.assertRaises(errors.TokenMismatch,
180
new_lockable.lock_write, token=different_token)
182
self.lockable.unlock()
144
184
def test_lock_write_with_matching_token(self):
145
185
# If the token matches, so no exception is raised by lock_write.
146
186
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()
189
# This test does not apply, because this lockable refuses
192
# The same instance will accept a second lock_write if the specified
194
self.lockable.lock_write(token=token)
195
self.lockable.unlock()
196
# Calling lock_write on a new instance for the same lockable will
198
new_lockable = self.get_lockable()
199
new_lockable.lock_write(token=token)
200
new_lockable.unlock()
202
self.lockable.unlock()
162
204
def test_unlock_after_lock_write_with_token(self):
163
205
# If lock_write did not physically acquire the lock (because it was
164
206
# passed a token), then unlock should not physically release it.
165
207
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())
210
# This test does not apply, because this lockable refuses
213
new_lockable = self.get_lockable()
214
new_lockable.lock_write(token=token)
215
new_lockable.unlock()
216
self.assertTrue(self.lockable.get_physical_lock_status())
218
self.lockable.unlock()
176
220
def test_lock_write_with_token_fails_when_unlocked(self):
177
221
# Lock and unlock to get a superficially valid token. This mimics a