← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: Robert Collins
  • Date: 2005-10-17 11:56:54 UTC
  • mfrom: (1185.16.59)
  • Revision ID: robertc@robertcollins.net-20051017115654-662239e1587524a8
merge from martin.

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
31
32
 
32
33
 
33
34
class TestStores(object):
34
 
    """Mixin template class that provides some common tests for stores"""
35
35
 
36
36
    def check_content(self, store, fileid, value):
37
37
        f = store.get(fileid)
79
79
 
80
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
81
81
 
82
 
    def get_store(self, path=u'.'):
 
82
    def get_store(self, path='.'):
83
83
        t = LocalTransport(path)
84
 
        return TextStore(t, compressed=True)
 
84
        return CompressedTextStore(t)
85
85
 
86
86
    def test_total_size(self):
87
 
        store = self.get_store(u'.')
 
87
        store = self.get_store('.')
88
88
        store.register_suffix('dsc')
89
89
        store.add(StringIO('goodbye'), '123123')
90
90
        store.add(StringIO('goodbye2'), '123123', 'dsc')
92
92
        self.assertEqual(store.total_size(), (2, 55))
93
93
        
94
94
    def test__relpath_suffixed(self):
95
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
95
        my_store = CompressedTextStore(MockTransport(), True)
96
96
        my_store.register_suffix('dsc')
97
 
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
97
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
98
98
 
99
99
 
100
100
class TestMemoryStore(TestCase):
138
138
 
139
139
class TestTextStore(TestCaseInTempDir, TestStores):
140
140
 
141
 
    def get_store(self, path=u'.'):
 
141
    def get_store(self, path='.'):
142
142
        t = LocalTransport(path)
143
 
        return TextStore(t, compressed=False)
 
143
        return TextStore(t)
144
144
 
145
145
    def test_total_size(self):
146
146
        store = self.get_store()
154
154
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
155
155
 
156
156
 
157
 
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
 
 
159
 
    def get_store(self, path=u'.', compressed=True):
160
 
        t = LocalTransport(path)
161
 
        return TextStore(t, compressed=compressed)
162
 
 
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')
167
 
 
168
 
        self.failUnlessExists('a.gz')
169
 
        self.failIf(os.path.lexists('a'))
170
 
 
171
 
        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
172
 
 
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')
177
 
        
178
 
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
179
 
 
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')
184
 
 
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')
189
 
        
190
 
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
191
 
 
192
157
class MockTransport(transport.Transport):
193
158
    """A fake transport for testing with."""
194
159
 
266
231
 
267
232
    def test__relpath_simple_suffixed(self):
268
233
        my_store = store.TransportStore(MockTransport())
 
234
        my_store.register_suffix('gz')
269
235
        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']))
 
236
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
237
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
273
238
 
274
239
    def test__relpath_prefixed_suffixed(self):
275
240
        my_store = store.TransportStore(MockTransport(), True)
 
241
        my_store.register_suffix('gz')
276
242
        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']))
 
243
        self.assertEqual('45/foo.gz', my_store._relpath('foo', ['gz']))
 
244
        self.assertEqual('45/foo.gz.bar',
 
245
                         my_store._relpath('foo', ['gz', 'bar']))
281
246
 
282
247
    def test_add_simple(self):
283
248
        stream = StringIO("content")
305
270
        my_store.add(stream, "foo", 'dsc')
306
271
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
307
272
 
308
 
    def get_populated_store(self, prefixed=False,
309
 
            store_class=TextStore, compressed=False):
310
 
        my_store = store_class(MemoryTransport(), prefixed,
311
 
                               compressed=compressed)
 
273
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
274
        my_store = store_class(MemoryTransport(), prefixed)
312
275
        my_store.register_suffix('sig')
313
276
        stream = StringIO("signature")
314
277
        my_store.add(stream, "foo", 'sig')
357
320
                         my_store.get('missing', 'sig').read())
358
321
 
359
322
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
323
        my_store = TextStore(MemoryTransport(), False)
361
324
        stream = StringIO("content")
362
325
        my_store.add(stream, "foo")
363
326
        self.assertEqual(set(['foo']),
372
335
    def test___iter__compressed(self):
373
336
        self.assertEqual(set(['foo']),
374
337
                         set(self.get_populated_store(
375
 
                             compressed=True).__iter__()))
 
338
                             store_class=CompressedTextStore).__iter__()))
376
339
        self.assertEqual(set(['foo']),
377
340
                         set(self.get_populated_store(
378
 
                             True, compressed=True).__iter__()))
 
341
                             True, CompressedTextStore).__iter__()))
379
342
 
380
343
    def test___len__(self):
381
344
        self.assertEqual(1, len(self.get_populated_store()))
382
345
 
383
346
    def test_copy_suffixes(self):
384
347
        from_store = self.get_populated_store()
385
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
348
        to_store = CompressedTextStore(MemoryTransport(), True)
386
349
        to_store.register_suffix('sig')
387
350
        copy_all(from_store, to_store)
388
351
        self.assertEqual(1, len(to_store))
390
353
        self.assertEqual('content', to_store.get('foo').read())
391
354
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
392
355
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
393
 
 
394
 
    def test_relpath_escaped(self):
395
 
        my_store = store.TransportStore(MemoryTransport())
396
 
        self.assertEqual('%25', my_store._relpath('%'))