~bzr-pqm/bzr/bzr.dev
» Revision
1185.12.49
(compared to revision 1185.16.85)
: bzrlib/selftest/teststore.py
« back to all changes in this revision
Viewing changes to bzrlib/selftest/teststore.py
- browse files at revision 1185.12.49
- compare with another revision
- download diff
- download tarball
- view history from revision 1185.12.49
- Committer: Aaron Bentley
- Date: 2005-10-18 18:48:27 UTC
- mto: (1185.25.1)
- mto: This revision was merged to the branch mainline in revision 1474.
- Revision ID: abentley@panoramicfeedback.com-20051018184827-2cc69376beb1cdf3
Switched to ConfigObj
- files added:
- bzrlib/util/configobj
- bzrlib/util/configobj/docs
- files removed:
- INSTALL
- files renamed:
- bzrlib/selftest/testplugins.py => bzrlib/selftest/plugins.py
- files modified:
- HACKING
- setup.py *
added
removed
bzrlib/selftest/teststore.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
16
16
17
"""Test Store implementations."""
18
17
"""Test Store implementation
18
"""
19
19
from cStringIO import StringIO
20
20
import os
21
21
22
from bzrlib.errors import BzrError, UnlistableStore
23
22
from bzrlib.store import copy_all
24
23
from bzrlib.transport.local import LocalTransport
25
24
from bzrlib.transport import NoSuchFile
26
25
from bzrlib.store.compressed_text import CompressedTextStore
27
26
from bzrlib.store.text import TextStore
28
27
from bzrlib.selftest import TestCase, TestCaseInTempDir
29
import bzrlib.store as store
30
import bzrlib.transport as transport
31
from bzrlib.transport.memory import MemoryTransport
32
33
34
class TestStores(object):
35
36
def check_content(self, store, fileid, value):
37
f = store.get(fileid)
38
self.assertEqual(f.read(), value)
39
40
def fill_store(self, store):
41
store.add(StringIO('hello'), 'a')
42
store.add(StringIO('other'), 'b')
43
store.add(StringIO('something'), 'c')
44
store.add(StringIO('goodbye'), '123123')
45
46
def test_copy_all(self):
47
"""Test copying"""
48
os.mkdir('a')
49
store_a = self.get_store('a')
50
store_a.add('foo', '1')
51
os.mkdir('b')
52
store_b = self.get_store('b')
53
copy_all(store_a, store_b)
54
self.assertEqual(store_a.get('1').read(), 'foo')
55
self.assertEqual(store_b.get('1').read(), 'foo')
56
# TODO: Switch the exception form UnlistableStore to
57
# or make Stores throw UnlistableStore if their
58
# Transport doesn't support listing
59
# store_c = RemoteStore('http://example.com/')
60
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
61
62
def test_get(self):
63
store = self.get_store()
64
self.fill_store(store)
65
66
self.check_content(store, 'a', 'hello')
67
self.check_content(store, 'b', 'other')
68
self.check_content(store, 'c', 'something')
69
70
# Make sure that requesting a non-existing file fails
71
self.assertRaises(KeyError, self.check_content, store, 'd', None)
28
from bzrlib.errors import BzrError, UnlistableStore
29
import bzrlib.store
30
31
32
def fill_store(store):
33
store.add(StringIO('hello'), 'a')
34
store.add(StringIO('other'), 'b')
35
store.add(StringIO('something'), 'c')
36
store.add(StringIO('goodbye'), '123123')
37
38
def check_equals(tester, store, files, values, permit_failure=False):
39
files = store.get(files, permit_failure=permit_failure)
40
count = 0
41
for f, v in zip(files, values):
42
count += 1
43
if v is None:
44
tester.assert_(f is None)
45
else:
46
tester.assertEquals(f.read(), v)
47
tester.assertEquals(count, len(values))
48
# We need to check to make sure there are no more
49
# files to be returned, I'm using a cheezy way
50
# Convert to a list, and there shouldn't be any left
51
tester.assertEquals(len(list(files)), 0)
52
53
def test_multiple_add(tester, store):
54
fill_store(store)
55
tester.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
56
57
def test_get(tester, store):
58
fill_store(store)
59
60
check_equals(tester, store, ['a'], ['hello'])
61
check_equals(tester, store, ['b', 'c'], ['other', 'something'])
62
63
# Make sure that requesting a non-existing file fails
64
tester.assertRaises(NoSuchFile, check_equals, tester, store,
65
['d'], [None])
66
tester.assertRaises(NoSuchFile, check_equals, tester, store,
67
['a', 'd'], ['hello', None])
68
tester.assertRaises(NoSuchFile, check_equals, tester, store,
69
['d', 'a'], [None, 'hello'])
70
tester.assertRaises(NoSuchFile, check_equals, tester, store,
71
['d', 'd', 'd'], [None, None, None])
72
tester.assertRaises(NoSuchFile, check_equals, tester, store,
73
['a', 'd', 'b'], ['hello', None, 'other'])
74
75
def test_ignore_get(tester, store):
76
fill_store(store)
77
78
files = store.get(['d'], permit_failure=True)
79
files = list(files)
80
tester.assertEquals(len(files), 1)
81
tester.assert_(files[0] is None)
82
83
check_equals(tester, store, ['a', 'd'], ['hello', None],
84
permit_failure=True)
85
check_equals(tester, store, ['d', 'a'], [None, 'hello'],
86
permit_failure=True)
87
check_equals(tester, store, ['d', 'd'], [None, None],
88
permit_failure=True)
89
check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
90
permit_failure=True)
91
check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
92
permit_failure=True)
93
check_equals(tester, store, ['b', 'd', 'c'], ['other', None, 'something'],
94
permit_failure=True)
95
96
97
def get_compressed_store(path='.'):
98
t = LocalTransport(path)
99
return CompressedTextStore(t)
100
101
102
def get_text_store(path='.'):
103
t = LocalTransport(path)
104
return TextStore(t)
105
106
107
class TestCompressedTextStore(TestCaseInTempDir):
72
108
73
109
def test_multiple_add(self):
74
110
"""Multiple add with same ID should raise a BzrError"""
75
store = self.get_store()
76
self.fill_store(store)
77
self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
78
79
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
81
82
def get_store(self, path='.'):
83
t = LocalTransport(path)
84
return CompressedTextStore(t)
111
store = get_compressed_store()
112
test_multiple_add(self, store)
113
114
def test_get(self):
115
store = get_compressed_store()
116
test_get(self, store)
117
118
def test_ignore_get(self):
119
store = get_compressed_store()
120
test_ignore_get(self, store)
85
121
86
122
def test_total_size(self):
87
store = self.get_store('.')
88
store.register_suffix('dsc')
123
store = get_compressed_store('.')
89
124
store.add(StringIO('goodbye'), '123123')
90
store.add(StringIO('goodbye2'), '123123', 'dsc')
125
store.add(StringIO('goodbye2'), '123123.dsc')
91
126
# these get gzipped - content should be stable
92
127
self.assertEqual(store.total_size(), (2, 55))
93
128
94
def test__relpath_suffixed(self):
95
my_store = CompressedTextStore(MockTransport(), True)
96
my_store.register_suffix('dsc')
97
self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
129
def test_copy_all(self):
130
"""Test copying"""
131
os.mkdir('a')
132
store_a = get_text_store('a')
133
store_a.add('foo', '1')
134
os.mkdir('b')
135
store_b = get_text_store('b')
136
copy_all(store_a, store_b)
137
self.assertEqual(store_a['1'].read(), 'foo')
138
self.assertEqual(store_b['1'].read(), 'foo')
98
139
99
140
100
141
class TestMemoryStore(TestCase):
101
142
102
143
def get_store(self):
103
return store.ImmutableMemoryStore()
144
return bzrlib.store.ImmutableMemoryStore()
104
145
105
146
def test_imports(self):
106
147
from bzrlib.store import ImmutableMemoryStore
108
149
def test_add_and_retrieve(self):
109
150
store = self.get_store()
110
151
store.add(StringIO('hello'), 'aa')
111
self.assertNotEqual(store.get('aa'), None)
112
self.assertEqual(store.get('aa').read(), 'hello')
152
self.assertNotEqual(store['aa'], None)
153
self.assertEqual(store['aa'].read(), 'hello')
113
154
store.add(StringIO('hello world'), 'bb')
114
self.assertNotEqual(store.get('bb'), None)
115
self.assertEqual(store.get('bb').read(), 'hello world')
155
self.assertNotEqual(store['bb'], None)
156
self.assertEqual(store['bb'].read(), 'hello world')
116
157
117
158
def test_missing_is_absent(self):
118
159
store = self.get_store()
119
160
self.failIf('aa' in store)
120
161
121
162
def test_adding_fails_when_present(self):
122
my_store = self.get_store()
123
my_store.add(StringIO('hello'), 'aa')
124
self.assertRaises(BzrError,
125
my_store.add, StringIO('hello'), 'aa')
126
127
def test_total_size(self):
128
store = self.get_store()
129
store.add(StringIO('goodbye'), '123123')
130
store.add(StringIO('goodbye2'), '123123.dsc')
131
self.assertEqual(store.total_size(), (2, 15))
132
# TODO: Switch the exception form UnlistableStore to
133
# or make Stores throw UnlistableStore if their
134
# Transport doesn't support listing
135
# store_c = RemoteStore('http://example.com/')
136
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
137
138
139
class TestTextStore(TestCaseInTempDir, TestStores):
140
141
def get_store(self, path='.'):
142
t = LocalTransport(path)
143
return TextStore(t)
144
145
def test_total_size(self):
146
store = self.get_store()
147
store.add(StringIO('goodbye'), '123123')
148
store.add(StringIO('goodbye2'), '123123.dsc')
149
self.assertEqual(store.total_size(), (2, 15))
150
# TODO: Switch the exception form UnlistableStore to
151
# or make Stores throw UnlistableStore if their
152
# Transport doesn't support listing
153
# store_c = RemoteStore('http://example.com/')
154
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
155
156
157
class MockTransport(transport.Transport):
158
"""A fake transport for testing with."""
159
160
def has(self, filename):
161
return False
162
163
def __init__(self, url=None):
164
if url is None:
165
url = "http://example.com"
166
super(MockTransport, self).__init__(url)
167
168
def mkdir(self, filename):
169
return
170
171
172
class InstrumentedTransportStore(store.TransportStore):
173
"""An instrumented TransportStore.
174
175
Here we replace template method worker methods with calls that record the
176
expected results.
177
"""
178
179
def _add(self, filename, file):
180
self._calls.append(("_add", filename, file))
181
182
def __init__(self, transport, prefixed=False):
183
super(InstrumentedTransportStore, self).__init__(transport, prefixed)
184
self._calls = []
185
186
187
class TestInstrumentedTransportStore(TestCase):
188
189
def test__add_records(self):
190
my_store = InstrumentedTransportStore(MockTransport())
191
my_store._add("filename", "file")
192
self.assertEqual([("_add", "filename", "file")], my_store._calls)
193
194
195
class TestMockTransport(TestCase):
196
197
def test_isinstance(self):
198
self.failUnless(isinstance(MockTransport(), transport.Transport))
199
200
def test_has(self):
201
self.assertEqual(False, MockTransport().has('foo'))
202
203
def test_mkdir(self):
204
MockTransport().mkdir('45')
205
206
207
class TestTransportStore(TestCase):
208
209
def test__relpath_invalid(self):
210
my_store = store.TransportStore(MockTransport())
211
self.assertRaises(ValueError, my_store._relpath, '/foo')
212
self.assertRaises(ValueError, my_store._relpath, 'foo/')
213
214
def test_register_invalid_suffixes(self):
215
my_store = store.TransportStore(MockTransport())
216
self.assertRaises(ValueError, my_store.register_suffix, '/')
217
self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
218
219
def test__relpath_unregister_suffixes(self):
220
my_store = store.TransportStore(MockTransport())
221
self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
222
self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
223
224
def test__relpath_simple(self):
225
my_store = store.TransportStore(MockTransport())
226
self.assertEqual("foo", my_store._relpath('foo'))
227
228
def test__relpath_prefixed(self):
229
my_store = store.TransportStore(MockTransport(), True)
230
self.assertEqual('45/foo', my_store._relpath('foo'))
231
232
def test__relpath_simple_suffixed(self):
233
my_store = store.TransportStore(MockTransport())
234
my_store.register_suffix('gz')
235
my_store.register_suffix('bar')
236
self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
237
self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
238
239
def test__relpath_prefixed_suffixed(self):
240
my_store = store.TransportStore(MockTransport(), True)
241
my_store.register_suffix('gz')
242
my_store.register_suffix('bar')
243
self.assertEqual('45/foo.gz', my_store._relpath('foo', ['gz']))
244
self.assertEqual('45/foo.gz.bar',
245
my_store._relpath('foo', ['gz', 'bar']))
246
247
def test_add_simple(self):
248
stream = StringIO("content")
249
my_store = InstrumentedTransportStore(MockTransport())
250
my_store.add(stream, "foo")
251
self.assertEqual([("_add", "foo", stream)], my_store._calls)
252
253
def test_add_prefixed(self):
254
stream = StringIO("content")
255
my_store = InstrumentedTransportStore(MockTransport(), True)
256
my_store.add(stream, "foo")
257
self.assertEqual([("_add", "45/foo", stream)], my_store._calls)
258
259
def test_add_simple_suffixed(self):
260
stream = StringIO("content")
261
my_store = InstrumentedTransportStore(MockTransport())
262
my_store.register_suffix('dsc')
263
my_store.add(stream, "foo", 'dsc')
264
self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
265
266
def test_add_simple_suffixed(self):
267
stream = StringIO("content")
268
my_store = InstrumentedTransportStore(MockTransport(), True)
269
my_store.register_suffix('dsc')
270
my_store.add(stream, "foo", 'dsc')
271
self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
272
273
def get_populated_store(self, prefixed=False, store_class=TextStore):
274
my_store = store_class(MemoryTransport(), prefixed)
275
my_store.register_suffix('sig')
276
stream = StringIO("signature")
277
my_store.add(stream, "foo", 'sig')
278
stream = StringIO("content")
279
my_store.add(stream, "foo")
280
stream = StringIO("signature for missing base")
281
my_store.add(stream, "missing", 'sig')
282
return my_store
283
284
def test_has_simple(self):
285
my_store = self.get_populated_store()
286
self.assertEqual(True, my_store.has_id('foo'))
287
my_store = self.get_populated_store(True)
288
self.assertEqual(True, my_store.has_id('foo'))
289
290
def test_has_suffixed(self):
291
my_store = self.get_populated_store()
292
self.assertEqual(True, my_store.has_id('foo', 'sig'))
293
my_store = self.get_populated_store(True)
294
self.assertEqual(True, my_store.has_id('foo', 'sig'))
295
296
def test_has_suffixed_no_base(self):
297
my_store = self.get_populated_store()
298
self.assertEqual(False, my_store.has_id('missing'))
299
my_store = self.get_populated_store(True)
300
self.assertEqual(False, my_store.has_id('missing'))
301
302
def test_get_simple(self):
303
my_store = self.get_populated_store()
304
self.assertEqual('content', my_store.get('foo').read())
305
my_store = self.get_populated_store(True)
306
self.assertEqual('content', my_store.get('foo').read())
307
308
def test_get_suffixed(self):
309
my_store = self.get_populated_store()
310
self.assertEqual('signature', my_store.get('foo', 'sig').read())
311
my_store = self.get_populated_store(True)
312
self.assertEqual('signature', my_store.get('foo', 'sig').read())
313
314
def test_get_suffixed_no_base(self):
315
my_store = self.get_populated_store()
316
self.assertEqual('signature for missing base',
317
my_store.get('missing', 'sig').read())
318
my_store = self.get_populated_store(True)
319
self.assertEqual('signature for missing base',
320
my_store.get('missing', 'sig').read())
321
322
def test___iter__no_suffix(self):
323
my_store = TextStore(MemoryTransport(), False)
324
stream = StringIO("content")
325
my_store.add(stream, "foo")
326
self.assertEqual(set(['foo']),
327
set(my_store.__iter__()))
328
329
def test___iter__(self):
330
self.assertEqual(set(['foo']),
331
set(self.get_populated_store().__iter__()))
332
self.assertEqual(set(['foo']),
333
set(self.get_populated_store(True).__iter__()))
334
335
def test___iter__compressed(self):
336
self.assertEqual(set(['foo']),
337
set(self.get_populated_store(
338
store_class=CompressedTextStore).__iter__()))
339
self.assertEqual(set(['foo']),
340
set(self.get_populated_store(
341
True, CompressedTextStore).__iter__()))
342
343
def test___len__(self):
344
self.assertEqual(1, len(self.get_populated_store()))
345
346
def test_copy_suffixes(self):
347
from_store = self.get_populated_store()
348
to_store = CompressedTextStore(MemoryTransport(), True)
349
to_store.register_suffix('sig')
350
copy_all(from_store, to_store)
351
self.assertEqual(1, len(to_store))
352
self.assertEqual(set(['foo']), set(to_store.__iter__()))
353
self.assertEqual('content', to_store.get('foo').read())
354
self.assertEqual('signature', to_store.get('foo', 'sig').read())
355
self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
356
357
def test_relpath_escaped(self):
358
my_store = store.TransportStore(MemoryTransport())
359
self.assertEqual('%25', my_store._relpath('%'))
163
store = self.get_store()
164
store.add(StringIO('hello'), 'aa')
165
self.assertRaises(bzrlib.store.StoreError,
166
store.add, StringIO('hello'), 'aa')
167
168
def test_total_size(self):
169
store = self.get_store()
170
store.add(StringIO('goodbye'), '123123')
171
store.add(StringIO('goodbye2'), '123123.dsc')
172
self.assertEqual(store.total_size(), (2, 15))
173
# TODO: Switch the exception form UnlistableStore to
174
# or make Stores throw UnlistableStore if their
175
# Transport doesn't support listing
176
# store_c = RemoteStore('http://example.com/')
177
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
178
179
180
class TestTextStore(TestCaseInTempDir):
181
def test_multiple_add(self):
182
"""Multiple add with same ID should raise a BzrError"""
183
store = get_text_store()
184
test_multiple_add(self, store)
185
186
def test_get(self):
187
store = get_text_store()
188
test_get(self, store)
189
190
def test_ignore_get(self):
191
store = get_text_store()
192
test_ignore_get(self, store)
193
194
def test_copy_all(self):
195
"""Test copying"""
196
os.mkdir('a')
197
store_a = get_text_store('a')
198
store_a.add('foo', '1')
199
os.mkdir('b')
200
store_b = get_text_store('b')
201
copy_all(store_a, store_b)
202
self.assertEqual(store_a['1'].read(), 'foo')
203
self.assertEqual(store_b['1'].read(), 'foo')
204
# TODO: Switch the exception form UnlistableStore to
205
# or make Stores throw UnlistableStore if their
206
# Transport doesn't support listing
207
# store_c = RemoteStore('http://example.com/')
208
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1