1
# Copyright (C) 2005, 2007 Canonical Ltd
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.
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.
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Test Store implementations."""
19
from cStringIO import StringIO
23
import bzrlib.errors as errors
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
25
from bzrlib.transport.local import LocalTransport
26
from bzrlib.store.text import TextStore
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
import bzrlib.store as store
29
import bzrlib.store.versioned
30
import bzrlib.transactions as transactions
31
import bzrlib.transport as transport
32
from bzrlib.transport.memory import MemoryTransport
35
class TestStores(object):
36
"""Mixin template class that provides some common tests for stores"""
38
def check_content(self, store, fileid, value):
40
self.assertEqual(f.read(), value)
42
def fill_store(self, store):
43
store.add(StringIO('hello'), 'a')
44
store.add(StringIO('other'), 'b')
45
store.add(StringIO('something'), 'c')
46
store.add(StringIO('goodbye'), '123123')
48
def test_copy_all(self):
51
store_a = self.get_store('a')
52
store_a.add(StringIO('foo'), '1')
54
store_b = self.get_store('b')
55
store_b.copy_all_ids(store_a)
56
self.assertEqual(store_a.get('1').read(), 'foo')
57
self.assertEqual(store_b.get('1').read(), 'foo')
58
# TODO: Switch the exception form UnlistableStore to
59
# or make Stores throw UnlistableStore if their
60
# Transport doesn't support listing
61
# store_c = RemoteStore('http://example.com/')
62
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
65
store = self.get_store()
66
self.fill_store(store)
68
self.check_content(store, 'a', 'hello')
69
self.check_content(store, 'b', 'other')
70
self.check_content(store, 'c', 'something')
72
# Make sure that requesting a non-existing file fails
73
self.assertRaises(KeyError, self.check_content, store, 'd', None)
75
def test_multiple_add(self):
76
"""Multiple add with same ID should raise a BzrError"""
77
store = self.get_store()
78
self.fill_store(store)
79
self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
82
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
84
def get_store(self, path=u'.'):
85
t = transport.get_transport(path)
86
return TextStore(t, compressed=True)
88
def test_total_size(self):
89
store = self.get_store(u'.')
90
store.register_suffix('dsc')
91
store.add(StringIO('goodbye'), '123123')
92
store.add(StringIO('goodbye2'), '123123', 'dsc')
93
# these get gzipped - content should be stable
94
self.assertEqual(store.total_size(), (2, 55))
96
def test__relpath_suffixed(self):
97
my_store = TextStore(MockTransport(),
98
prefixed=True, compressed=True)
99
my_store.register_suffix('dsc')
100
self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
103
class TestMemoryStore(TestCase):
106
return TextStore(MemoryTransport())
108
def test_add_and_retrieve(self):
109
store = self.get_store()
110
store.add(StringIO('hello'), 'aa')
111
self.assertNotEqual(store.get('aa'), None)
112
self.assertEqual(store.get('aa').read(), 'hello')
113
store.add(StringIO('hello world'), 'bb')
114
self.assertNotEqual(store.get('bb'), None)
115
self.assertEqual(store.get('bb').read(), 'hello world')
117
def test_missing_is_absent(self):
118
store = self.get_store()
119
self.failIf('aa' in store)
121
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')
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)
139
class TestTextStore(TestCaseInTempDir, TestStores):
141
def get_store(self, path=u'.'):
142
t = transport.get_transport(path)
143
return TextStore(t, compressed=False)
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)
157
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
def get_store(self, path=u'.', compressed=True):
160
t = transport.get_transport(path)
161
return TextStore(t, compressed=compressed)
163
def test_get_mixed(self):
164
cs = self.get_store(u'.', compressed=True)
165
s = self.get_store(u'.', compressed=False)
166
cs.add(StringIO('hello there'), 'a')
168
self.failUnlessExists('a.gz')
169
self.failIf(os.path.lexists('a'))
171
self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
173
self.assertEquals(cs.has_id('a'), True)
174
self.assertEquals(s.has_id('a'), True)
175
self.assertEquals(cs.get('a').read(), 'hello there')
176
self.assertEquals(s.get('a').read(), 'hello there')
178
self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
180
s.add(StringIO('goodbye'), 'b')
181
self.failUnlessExists('b')
182
self.failIf(os.path.lexists('b.gz'))
183
self.assertEquals(open('b').read(), 'goodbye')
185
self.assertEquals(cs.has_id('b'), True)
186
self.assertEquals(s.has_id('b'), True)
187
self.assertEquals(cs.get('b').read(), 'goodbye')
188
self.assertEquals(s.get('b').read(), 'goodbye')
190
self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
192
class MockTransport(transport.Transport):
193
"""A fake transport for testing with."""
195
def has(self, filename):
198
def __init__(self, url=None):
200
url = "http://example.com"
201
super(MockTransport, self).__init__(url)
203
def mkdir(self, filename):
207
class InstrumentedTransportStore(store.TransportStore):
208
"""An instrumented TransportStore.
210
Here we replace template method worker methods with calls that record the
214
def _add(self, filename, file):
215
self._calls.append(("_add", filename, file))
217
def __init__(self, transport, prefixed=False):
218
super(InstrumentedTransportStore, self).__init__(transport, prefixed)
222
class TestInstrumentedTransportStore(TestCase):
224
def test__add_records(self):
225
my_store = InstrumentedTransportStore(MockTransport())
226
my_store._add("filename", "file")
227
self.assertEqual([("_add", "filename", "file")], my_store._calls)
230
class TestMockTransport(TestCase):
232
def test_isinstance(self):
233
self.failUnless(isinstance(MockTransport(), transport.Transport))
236
self.assertEqual(False, MockTransport().has('foo'))
238
def test_mkdir(self):
239
MockTransport().mkdir('45')
242
class TestTransportStore(TestCase):
244
def test__relpath_invalid(self):
245
my_store = store.TransportStore(MockTransport())
246
self.assertRaises(ValueError, my_store._relpath, '/foo')
247
self.assertRaises(ValueError, my_store._relpath, 'foo/')
249
def test_register_invalid_suffixes(self):
250
my_store = store.TransportStore(MockTransport())
251
self.assertRaises(ValueError, my_store.register_suffix, '/')
252
self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
254
def test__relpath_unregister_suffixes(self):
255
my_store = store.TransportStore(MockTransport())
256
self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
257
self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
259
def test__relpath_simple(self):
260
my_store = store.TransportStore(MockTransport())
261
self.assertEqual("foo", my_store._relpath('foo'))
263
def test__relpath_prefixed(self):
264
my_store = store.TransportStore(MockTransport(), True)
265
self.assertEqual('45/foo', my_store._relpath('foo'))
267
def test__relpath_simple_suffixed(self):
268
my_store = store.TransportStore(MockTransport())
269
my_store.register_suffix('bar')
270
my_store.register_suffix('baz')
271
self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
272
self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
274
def test__relpath_prefixed_suffixed(self):
275
my_store = store.TransportStore(MockTransport(), True)
276
my_store.register_suffix('bar')
277
my_store.register_suffix('baz')
278
self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
279
self.assertEqual('45/foo.bar.baz',
280
my_store._relpath('foo', ['bar', 'baz']))
282
def test_add_simple(self):
283
stream = StringIO("content")
284
my_store = InstrumentedTransportStore(MockTransport())
285
my_store.add(stream, "foo")
286
self.assertEqual([("_add", "foo", stream)], my_store._calls)
288
def test_add_prefixed(self):
289
stream = StringIO("content")
290
my_store = InstrumentedTransportStore(MockTransport(), True)
291
my_store.add(stream, "foo")
292
self.assertEqual([("_add", "45/foo", stream)], my_store._calls)
294
def test_add_simple_suffixed(self):
295
stream = StringIO("content")
296
my_store = InstrumentedTransportStore(MockTransport())
297
my_store.register_suffix('dsc')
298
my_store.add(stream, "foo", 'dsc')
299
self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
301
def test_add_simple_suffixed(self):
302
stream = StringIO("content")
303
my_store = InstrumentedTransportStore(MockTransport(), True)
304
my_store.register_suffix('dsc')
305
my_store.add(stream, "foo", 'dsc')
306
self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
308
def get_populated_store(self, prefixed=False,
309
store_class=TextStore, compressed=False):
310
my_store = store_class(MemoryTransport(), prefixed,
311
compressed=compressed)
312
my_store.register_suffix('sig')
313
stream = StringIO("signature")
314
my_store.add(stream, "foo", 'sig')
315
stream = StringIO("content")
316
my_store.add(stream, "foo")
317
stream = StringIO("signature for missing base")
318
my_store.add(stream, "missing", 'sig')
321
def test_has_simple(self):
322
my_store = self.get_populated_store()
323
self.assertEqual(True, my_store.has_id('foo'))
324
my_store = self.get_populated_store(True)
325
self.assertEqual(True, my_store.has_id('foo'))
327
def test_has_suffixed(self):
328
my_store = self.get_populated_store()
329
self.assertEqual(True, my_store.has_id('foo', 'sig'))
330
my_store = self.get_populated_store(True)
331
self.assertEqual(True, my_store.has_id('foo', 'sig'))
333
def test_has_suffixed_no_base(self):
334
my_store = self.get_populated_store()
335
self.assertEqual(False, my_store.has_id('missing'))
336
my_store = self.get_populated_store(True)
337
self.assertEqual(False, my_store.has_id('missing'))
339
def test_get_simple(self):
340
my_store = self.get_populated_store()
341
self.assertEqual('content', my_store.get('foo').read())
342
my_store = self.get_populated_store(True)
343
self.assertEqual('content', my_store.get('foo').read())
345
def test_get_suffixed(self):
346
my_store = self.get_populated_store()
347
self.assertEqual('signature', my_store.get('foo', 'sig').read())
348
my_store = self.get_populated_store(True)
349
self.assertEqual('signature', my_store.get('foo', 'sig').read())
351
def test_get_suffixed_no_base(self):
352
my_store = self.get_populated_store()
353
self.assertEqual('signature for missing base',
354
my_store.get('missing', 'sig').read())
355
my_store = self.get_populated_store(True)
356
self.assertEqual('signature for missing base',
357
my_store.get('missing', 'sig').read())
359
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(),
361
prefixed=False, compressed=False)
362
stream = StringIO("content")
363
my_store.add(stream, "foo")
364
self.assertEqual(set(['foo']),
365
set(my_store.__iter__()))
367
def test___iter__(self):
368
self.assertEqual(set(['foo']),
369
set(self.get_populated_store().__iter__()))
370
self.assertEqual(set(['foo']),
371
set(self.get_populated_store(True).__iter__()))
373
def test___iter__compressed(self):
374
self.assertEqual(set(['foo']),
375
set(self.get_populated_store(
376
compressed=True).__iter__()))
377
self.assertEqual(set(['foo']),
378
set(self.get_populated_store(
379
True, compressed=True).__iter__()))
381
def test___len__(self):
382
self.assertEqual(1, len(self.get_populated_store()))
384
def test_copy_suffixes(self):
385
from_store = self.get_populated_store()
386
to_store = TextStore(MemoryTransport(),
387
prefixed=True, compressed=True)
388
to_store.register_suffix('sig')
389
to_store.copy_all_ids(from_store)
390
self.assertEqual(1, len(to_store))
391
self.assertEqual(set(['foo']), set(to_store.__iter__()))
392
self.assertEqual('content', to_store.get('foo').read())
393
self.assertEqual('signature', to_store.get('foo', 'sig').read())
394
self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
396
def test_relpath_escaped(self):
397
my_store = store.TransportStore(MemoryTransport())
398
self.assertEqual('%25', my_store._relpath('%'))
400
def test_escaped_uppercase(self):
401
"""Uppercase letters are escaped for safety on Windows"""
402
my_store = store.TransportStore(MemoryTransport(), prefixed=True,
404
# a particularly perverse file-id! :-)
405
self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
408
class TestVersionFileStore(TestCaseWithTransport):
411
return self._transaction
414
super(TestVersionFileStore, self).setUp()
415
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
416
self.vfstore.get_scope = self.get_scope
417
self._transaction = None
419
def test_get_weave_registers_dirty_in_write(self):
420
self._transaction = transactions.WriteTransaction()
421
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
422
self._transaction.finish()
423
self._transaction = None
424
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
425
self._transaction = transactions.WriteTransaction()
426
vf = self.vfstore.get_weave('id', self._transaction)
427
self._transaction.finish()
428
self._transaction = None
429
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
431
def test_get_weave_readonly_cant_write(self):
432
self._transaction = transactions.WriteTransaction()
433
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
434
self._transaction.finish()
435
self._transaction = transactions.ReadOnlyTransaction()
436
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
437
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
439
def test___iter__escaped(self):
440
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
441
prefixed=True, escaped=True)
442
self.vfstore.get_scope = self.get_scope
443
self._transaction = transactions.WriteTransaction()
444
vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
445
vf.add_lines('a', [], [])
447
self._transaction.finish()
448
self.assertEqual([' '], list(self.vfstore))
added
removed
bzrlib/tests/test_store.py
