← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

Handled more pipe errors for display commands.

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
22
21
 
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
79
80
 
80
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
81
82
 
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)
85
86
 
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))
93
94
        
94
95
    def test__relpath_suffixed(self):
95
 
        my_store = TextStore(MockTransport(),
96
 
                             prefixed=True, compressed=True)
 
96
        my_store = CompressedTextStore(MockTransport(), True)
97
97
        my_store.register_suffix('dsc')
98
 
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
98
        self.assertEqual('45/foo.dsc.gz', 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=u'.'):
 
142
    def get_store(self, path='.'):
143
143
        t = LocalTransport(path)
144
 
        return TextStore(t, compressed=False)
 
144
        return TextStore(t)
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
 
 
193
158
class MockTransport(transport.Transport):
194
159
    """A fake transport for testing with."""
195
160
 
267
232
 
268
233
    def test__relpath_simple_suffixed(self):
269
234
        my_store = store.TransportStore(MockTransport())
 
235
        my_store.register_suffix('gz')
270
236
        my_store.register_suffix('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']))
 
237
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
238
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
274
239
 
275
240
    def test__relpath_prefixed_suffixed(self):
276
241
        my_store = store.TransportStore(MockTransport(), True)
 
242
        my_store.register_suffix('gz')
277
243
        my_store.register_suffix('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']))
 
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
247
 
283
248
    def test_add_simple(self):
284
249
        stream = StringIO("content")
306
271
        my_store.add(stream, "foo", 'dsc')
307
272
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
308
273
 
309
 
    def get_populated_store(self, prefixed=False,
310
 
            store_class=TextStore, compressed=False):
311
 
        my_store = store_class(MemoryTransport(), prefixed,
312
 
                               compressed=compressed)
 
274
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
275
        my_store = store_class(MemoryTransport(), prefixed)
313
276
        my_store.register_suffix('sig')
314
277
        stream = StringIO("signature")
315
278
        my_store.add(stream, "foo", 'sig')
358
321
                         my_store.get('missing', 'sig').read())
359
322
 
360
323
    def test___iter__no_suffix(self):
361
 
        my_store = TextStore(MemoryTransport(),
362
 
                             prefixed=False, compressed=False)
 
324
        my_store = TextStore(MemoryTransport(), False)
363
325
        stream = StringIO("content")
364
326
        my_store.add(stream, "foo")
365
327
        self.assertEqual(set(['foo']),
374
336
    def test___iter__compressed(self):
375
337
        self.assertEqual(set(['foo']),
376
338
                         set(self.get_populated_store(
377
 
                             compressed=True).__iter__()))
 
339
                             store_class=CompressedTextStore).__iter__()))
378
340
        self.assertEqual(set(['foo']),
379
341
                         set(self.get_populated_store(
380
 
                             True, compressed=True).__iter__()))
 
342
                             True, CompressedTextStore).__iter__()))
381
343
 
382
344
    def test___len__(self):
383
345
        self.assertEqual(1, len(self.get_populated_store()))
384
346
 
385
347
    def test_copy_suffixes(self):
386
348
        from_store = self.get_populated_store()
387
 
        to_store = TextStore(MemoryTransport(),
388
 
                             prefixed=True, compressed=True)
 
349
        to_store = CompressedTextStore(MemoryTransport(), True)
389
350
        to_store.register_suffix('sig')
390
351
        copy_all(from_store, to_store)
391
352
        self.assertEqual(1, len(to_store))