19
19
from cStringIO import StringIO
23
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
22
from bzrlib.errors import BzrError, UnlistableStore
24
23
from bzrlib.store import copy_all
25
24
from bzrlib.transport.local import LocalTransport
25
from bzrlib.transport import NoSuchFile
26
from bzrlib.store.compressed_text import CompressedTextStore
26
27
from bzrlib.store.text import TextStore
27
from bzrlib.tests import TestCase, TestCaseInTempDir
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
28
29
import bzrlib.store as store
29
30
import bzrlib.transport as transport
30
31
from bzrlib.transport.memory import MemoryTransport
80
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
def get_store(self, path=u'.'):
83
def get_store(self, path='.'):
83
84
t = LocalTransport(path)
84
return TextStore(t, compressed=True)
85
return CompressedTextStore(t)
86
87
def test_total_size(self):
87
store = self.get_store(u'.')
88
store = self.get_store('.')
88
89
store.register_suffix('dsc')
89
90
store.add(StringIO('goodbye'), '123123')
90
91
store.add(StringIO('goodbye2'), '123123', 'dsc')
92
93
self.assertEqual(store.total_size(), (2, 55))
94
95
def test__relpath_suffixed(self):
95
my_store = TextStore(MockTransport(), True, compressed=True)
96
my_store = CompressedTextStore(MockTransport(), True)
96
97
my_store.register_suffix('dsc')
97
self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
98
self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
100
101
class TestMemoryStore(TestCase):
139
140
class TestTextStore(TestCaseInTempDir, TestStores):
141
def get_store(self, path=u'.'):
142
def get_store(self, path='.'):
142
143
t = LocalTransport(path)
143
return TextStore(t, compressed=False)
145
146
def test_total_size(self):
146
147
store = self.get_store()
154
155
# 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 = LocalTransport(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
158
class MockTransport(transport.Transport):
193
159
"""A fake transport for testing with."""
267
233
def test__relpath_simple_suffixed(self):
268
234
my_store = store.TransportStore(MockTransport())
235
my_store.register_suffix('gz')
269
236
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']))
237
self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
238
self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
274
240
def test__relpath_prefixed_suffixed(self):
275
241
my_store = store.TransportStore(MockTransport(), True)
242
my_store.register_suffix('gz')
276
243
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']))
244
self.assertEqual('45/foo.gz', my_store._relpath('foo', ['gz']))
245
self.assertEqual('45/foo.gz.bar',
246
my_store._relpath('foo', ['gz', 'bar']))
282
248
def test_add_simple(self):
283
249
stream = StringIO("content")
305
271
my_store.add(stream, "foo", 'dsc')
306
272
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)
274
def get_populated_store(self, prefixed=False, store_class=TextStore):
275
my_store = store_class(MemoryTransport(), prefixed)
312
276
my_store.register_suffix('sig')
313
277
stream = StringIO("signature")
314
278
my_store.add(stream, "foo", 'sig')
357
321
my_store.get('missing', 'sig').read())
359
323
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(), False, compressed=False)
324
my_store = TextStore(MemoryTransport(), False)
361
325
stream = StringIO("content")
362
326
my_store.add(stream, "foo")
363
327
self.assertEqual(set(['foo']),
372
336
def test___iter__compressed(self):
373
337
self.assertEqual(set(['foo']),
374
338
set(self.get_populated_store(
375
compressed=True).__iter__()))
339
store_class=CompressedTextStore).__iter__()))
376
340
self.assertEqual(set(['foo']),
377
341
set(self.get_populated_store(
378
True, compressed=True).__iter__()))
342
True, CompressedTextStore).__iter__()))
380
344
def test___len__(self):
381
345
self.assertEqual(1, len(self.get_populated_store()))
383
347
def test_copy_suffixes(self):
384
348
from_store = self.get_populated_store()
385
to_store = TextStore(MemoryTransport(), True, compressed=True)
349
to_store = CompressedTextStore(MemoryTransport(), True)
386
350
to_store.register_suffix('sig')
387
351
copy_all(from_store, to_store)
388
352
self.assertEqual(1, len(to_store))