← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

[merge] robertc's integration, updated tests to check for retcode=3

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
18
18
 
19
19
from cStringIO import StringIO
20
20
import os
 
21
import gzip
21
22
 
22
23
from bzrlib.errors import BzrError, UnlistableStore
23
24
from bzrlib.store import copy_all
24
25
from bzrlib.transport.local import LocalTransport
25
26
from bzrlib.transport import NoSuchFile
26
 
from bzrlib.store.compressed_text import CompressedTextStore
27
27
from bzrlib.store.text import TextStore
28
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
29
29
import bzrlib.store as store
82
82
 
83
83
    def get_store(self, path='.'):
84
84
        t = LocalTransport(path)
85
 
        return CompressedTextStore(t)
 
85
        return TextStore(t, compressed=True)
86
86
 
87
87
    def test_total_size(self):
88
88
        store = self.get_store('.')
93
93
        self.assertEqual(store.total_size(), (2, 55))
94
94
        
95
95
    def test__relpath_suffixed(self):
96
 
        my_store = CompressedTextStore(MockTransport(), True)
 
96
        my_store = TextStore(MockTransport(), 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):
141
141
 
142
142
    def get_store(self, path='.'):
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='.', compressed=True):
 
161
        t = LocalTransport(path)
 
162
        return TextStore(t, compressed=compressed)
 
163
 
 
164
    def test_get_mixed(self):
 
165
        cs = self.get_store('.', compressed=True)
 
166
        s = self.get_store('.', 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(), False, compressed=False)
325
362
        stream = StringIO("content")
326
363
        my_store.add(stream, "foo")
327
364
        self.assertEqual(set(['foo']),
336
373
    def test___iter__compressed(self):
337
374
        self.assertEqual(set(['foo']),
338
375
                         set(self.get_populated_store(
339
 
                             store_class=CompressedTextStore).__iter__()))
 
376
                             compressed=True).__iter__()))
340
377
        self.assertEqual(set(['foo']),
341
378
                         set(self.get_populated_store(
342
 
                             True, CompressedTextStore).__iter__()))
 
379
                             True, compressed=True).__iter__()))
343
380
 
344
381
    def test___len__(self):
345
382
        self.assertEqual(1, len(self.get_populated_store()))
346
383
 
347
384
    def test_copy_suffixes(self):
348
385
        from_store = self.get_populated_store()
349
 
        to_store = CompressedTextStore(MemoryTransport(), True)
 
386
        to_store = TextStore(MemoryTransport(), True, compressed=True)
350
387
        to_store.register_suffix('sig')
351
388
        copy_all(from_store, to_store)
352
389
        self.assertEqual(1, len(to_store))