← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

[merge] jam-integration 1495

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
18
18
 
19
19
from cStringIO import StringIO
20
20
import os
 
21
import gzip
21
22
 
22
 
from bzrlib.errors import BzrError, UnlistableStore
 
23
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
23
24
from bzrlib.store import copy_all
24
25
from bzrlib.transport.local import LocalTransport
25
 
from bzrlib.transport import NoSuchFile
26
 
from bzrlib.store.compressed_text import CompressedTextStore
27
26
from bzrlib.store.text import TextStore
28
 
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir
29
28
import bzrlib.store as store
30
29
import bzrlib.transport as transport
31
30
from bzrlib.transport.memory import MemoryTransport
80
79
 
81
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
81
 
83
 
    def get_store(self, path='.'):
 
82
    def get_store(self, path=u'.'):
84
83
        t = LocalTransport(path)
85
 
        return CompressedTextStore(t)
 
84
        return TextStore(t, compressed=True)
86
85
 
87
86
    def test_total_size(self):
88
 
        store = self.get_store('.')
 
87
        store = self.get_store(u'.')
89
88
        store.register_suffix('dsc')
90
89
        store.add(StringIO('goodbye'), '123123')
91
90
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
92
        self.assertEqual(store.total_size(), (2, 55))
94
93
        
95
94
    def test__relpath_suffixed(self):
96
 
        my_store = CompressedTextStore(MockTransport(), True)
 
95
        my_store = TextStore(MockTransport(),
 
96
                             prefixed=True, compressed=True)
97
97
        my_store.register_suffix('dsc')
98
 
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
 
98
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
99
99
 
100
100
 
101
101
class TestMemoryStore(TestCase):
139
139
 
140
140
class TestTextStore(TestCaseInTempDir, TestStores):
141
141
 
142
 
    def get_store(self, path='.'):
 
142
    def get_store(self, path=u'.'):
143
143
        t = LocalTransport(path)
144
 
        return TextStore(t)
 
144
        return TextStore(t, compressed=False)
145
145
 
146
146
    def test_total_size(self):
147
147
        store = self.get_store()
155
155
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
156
156
 
157
157
 
 
158
class TestMixedTextStore(TestCaseInTempDir, TestStores):
 
159
 
 
160
    def get_store(self, path=u'.', compressed=True):
 
161
        t = LocalTransport(path)
 
162
        return TextStore(t, compressed=compressed)
 
163
 
 
164
    def test_get_mixed(self):
 
165
        cs = self.get_store(u'.', compressed=True)
 
166
        s = self.get_store(u'.', compressed=False)
 
167
        cs.add(StringIO('hello there'), 'a')
 
168
 
 
169
        self.failUnlessExists('a.gz')
 
170
        self.failIf(os.path.lexists('a'))
 
171
 
 
172
        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
 
173
 
 
174
        self.assertEquals(cs.has_id('a'), True)
 
175
        self.assertEquals(s.has_id('a'), True)
 
176
        self.assertEquals(cs.get('a').read(), 'hello there')
 
177
        self.assertEquals(s.get('a').read(), 'hello there')
 
178
        
 
179
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
 
180
 
 
181
        s.add(StringIO('goodbye'), 'b')
 
182
        self.failUnlessExists('b')
 
183
        self.failIf(os.path.lexists('b.gz'))
 
184
        self.assertEquals(open('b').read(), 'goodbye')
 
185
 
 
186
        self.assertEquals(cs.has_id('b'), True)
 
187
        self.assertEquals(s.has_id('b'), True)
 
188
        self.assertEquals(cs.get('b').read(), 'goodbye')
 
189
        self.assertEquals(s.get('b').read(), 'goodbye')
 
190
        
 
191
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
 
192
 
158
193
class MockTransport(transport.Transport):
159
194
    """A fake transport for testing with."""
160
195
 
232
267
 
233
268
    def test__relpath_simple_suffixed(self):
234
269
        my_store = store.TransportStore(MockTransport())
235
 
        my_store.register_suffix('gz')
236
270
        my_store.register_suffix('bar')
237
 
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
238
 
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
 
271
        my_store.register_suffix('baz')
 
272
        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
 
273
        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
239
274
 
240
275
    def test__relpath_prefixed_suffixed(self):
241
276
        my_store = store.TransportStore(MockTransport(), True)
242
 
        my_store.register_suffix('gz')
243
277
        my_store.register_suffix('bar')
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']))
 
278
        my_store.register_suffix('baz')
 
279
        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
 
280
        self.assertEqual('45/foo.bar.baz',
 
281
                         my_store._relpath('foo', ['bar', 'baz']))
247
282
 
248
283
    def test_add_simple(self):
249
284
        stream = StringIO("content")
271
306
        my_store.add(stream, "foo", 'dsc')
272
307
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
273
308
 
274
 
    def get_populated_store(self, prefixed=False, store_class=TextStore):
275
 
        my_store = store_class(MemoryTransport(), prefixed)
 
309
    def get_populated_store(self, prefixed=False,
 
310
            store_class=TextStore, compressed=False):
 
311
        my_store = store_class(MemoryTransport(), prefixed,
 
312
                               compressed=compressed)
276
313
        my_store.register_suffix('sig')
277
314
        stream = StringIO("signature")
278
315
        my_store.add(stream, "foo", 'sig')
321
358
                         my_store.get('missing', 'sig').read())
322
359
 
323
360
    def test___iter__no_suffix(self):
324
 
        my_store = TextStore(MemoryTransport(), False)
 
361
        my_store = TextStore(MemoryTransport(),
 
362
                             prefixed=False, compressed=False)
325
363
        stream = StringIO("content")
326
364
        my_store.add(stream, "foo")
327
365
        self.assertEqual(set(['foo']),
336
374
    def test___iter__compressed(self):
337
375
        self.assertEqual(set(['foo']),
338
376
                         set(self.get_populated_store(
339
 
                             store_class=CompressedTextStore).__iter__()))
 
377
                             compressed=True).__iter__()))
340
378
        self.assertEqual(set(['foo']),
341
379
                         set(self.get_populated_store(
342
 
                             True, CompressedTextStore).__iter__()))
 
380
                             True, compressed=True).__iter__()))
343
381
 
344
382
    def test___len__(self):
345
383
        self.assertEqual(1, len(self.get_populated_store()))
346
384
 
347
385
    def test_copy_suffixes(self):
348
386
        from_store = self.get_populated_store()
349
 
        to_store = CompressedTextStore(MemoryTransport(), True)
 
387
        to_store = TextStore(MemoryTransport(),
 
388
                             prefixed=True, compressed=True)
350
389
        to_store.register_suffix('sig')
351
390
        copy_all(from_store, to_store)
352
391
        self.assertEqual(1, len(to_store))