« back to all changes in this revision
Viewing changes to bzrlib/selftest/teststore.py
- browse files at revision 1185.16.72
- compare with another revision
- download diff
- download tarball
- view history from revision 1185.16.72
- Committer: Martin Pool
- Date: 2005-10-18 06:30:47 UTC
- mfrom: (1462)
- mto: (1185.41.1 bzr.sftp) (1464.1.1) (1534.1.1 integration) (1495.1.4) (1505.1.16 bzr-bound-branch) (1508.1.15) (1526.1.3 run_tests_twice_for_i18n) (1540.1.3 bzr.dev (Main development branch)) (1685.1.1 bzr-encoding) (1553.5.1 bzr.dev (Main development branch)) (1608.2.1 bzr.mbp.escape-stores)
- mto: This revision was merged to the branch mainline in revision 1464.
- Revision ID: mbp@sourcefrog.net-20051018063047-13fc3067d2f579a3
[merge] from robert and fix up tests
- files added:
- bzrlib/gpg.py
- files modified:
- NEWS
- 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 implementation
18
"""
17
"""Test Store implementations."""
18
19
19
from cStringIO import StringIO
20
20
import os
21
21
22
from bzrlib.errors import BzrError, UnlistableStore
22
23
from bzrlib.store import copy_all
23
24
from bzrlib.transport.local import LocalTransport
24
25
from bzrlib.transport import NoSuchFile
25
26
from bzrlib.store.compressed_text import CompressedTextStore
26
27
from bzrlib.store.text import TextStore
27
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
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):
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)
108
72
109
73
def test_multiple_add(self):
110
74
"""Multiple add with same ID should raise a BzrError"""
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)
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)
121
85
122
86
def test_total_size(self):
123
store = get_compressed_store('.')
87
store = self.get_store('.')
88
store.register_suffix('dsc')
124
89
store.add(StringIO('goodbye'), '123123')
125
store.add(StringIO('goodbye2'), '123123.dsc')
90
store.add(StringIO('goodbye2'), '123123', 'dsc')
126
91
# these get gzipped - content should be stable
127
92
self.assertEqual(store.total_size(), (2, 55))
128
93
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')
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']))
139
98
140
99
141
100
class TestMemoryStore(TestCase):
142
101
143
102
def get_store(self):
144
return bzrlib.store.ImmutableMemoryStore()
103
return store.ImmutableMemoryStore()
145
104
146
105
def test_imports(self):
147
106
from bzrlib.store import ImmutableMemoryStore
149
108
def test_add_and_retrieve(self):
150
109
store = self.get_store()
151
110
store.add(StringIO('hello'), 'aa')
152
self.assertNotEqual(store['aa'], None)
153
self.assertEqual(store['aa'].read(), 'hello')
111
self.assertNotEqual(store.get('aa'), None)
112
self.assertEqual(store.get('aa').read(), 'hello')
154
113
store.add(StringIO('hello world'), 'bb')
155
self.assertNotEqual(store['bb'], None)
156
self.assertEqual(store['bb'].read(), 'hello world')
114
self.assertNotEqual(store.get('bb'), None)
115
self.assertEqual(store.get('bb').read(), 'hello world')
157
116
158
117
def test_missing_is_absent(self):
159
118
store = self.get_store()
160
119
self.failIf('aa' in store)
161
120
162
121
def test_adding_fails_when_present(self):
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)
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')
Loggerhead is a web-based interface for Breezy
Version: 2.0.1