← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Michael Ellerman
  • Date: 2005-12-10 22:11:13 UTC
  • mto: This revision was merged to the branch mainline in revision 1528.
  • Revision ID: michael@ellerman.id.au-20051210221113-99ca561aaab4661e
Simplify handling of DivergedBranches in cmd_pull()

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
32
31
 
33
32
 
34
33
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='.'):
 
82
    def get_store(self, path=u'.'):
83
83
        t = LocalTransport(path)
84
 
        return CompressedTextStore(t)
 
84
        return TextStore(t, compressed=True)
85
85
 
86
86
    def test_total_size(self):
87
 
        store = self.get_store('.')
 
87
        store = self.get_store(u'.')
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 = CompressedTextStore(MockTransport(), True)
 
95
        my_store = TextStore(MockTransport(), True, compressed=True)
96
96
        my_store.register_suffix('dsc')
97
 
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
 
97
        self.assertEqual('45/foo.dsc', 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='.'):
 
141
    def get_store(self, path=u'.'):
142
142
        t = LocalTransport(path)
143
 
        return TextStore(t)
 
143
        return TextStore(t, compressed=False)
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
 
157
192
class MockTransport(transport.Transport):
158
193
    """A fake transport for testing with."""
159
194
 
231
266
 
232
267
    def test__relpath_simple_suffixed(self):
233
268
        my_store = store.TransportStore(MockTransport())
234
 
        my_store.register_suffix('gz')
235
269
        my_store.register_suffix('bar')
236
 
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
237
 
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', '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']))
238
273
 
239
274
    def test__relpath_prefixed_suffixed(self):
240
275
        my_store = store.TransportStore(MockTransport(), True)
241
 
        my_store.register_suffix('gz')
242
276
        my_store.register_suffix('bar')
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']))
 
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']))
246
281
 
247
282
    def test_add_simple(self):
248
283
        stream = StringIO("content")
270
305
        my_store.add(stream, "foo", 'dsc')
271
306
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
272
307
 
273
 
    def get_populated_store(self, prefixed=False, store_class=TextStore):
274
 
        my_store = store_class(MemoryTransport(), prefixed)
 
308
    def get_populated_store(self, prefixed=False,
 
309
            store_class=TextStore, compressed=False):
 
310
        my_store = store_class(MemoryTransport(), prefixed,
 
311
                               compressed=compressed)
275
312
        my_store.register_suffix('sig')
276
313
        stream = StringIO("signature")
277
314
        my_store.add(stream, "foo", 'sig')
320
357
                         my_store.get('missing', 'sig').read())
321
358
 
322
359
    def test___iter__no_suffix(self):
323
 
        my_store = TextStore(MemoryTransport(), False)
 
360
        my_store = TextStore(MemoryTransport(), False, compressed=False)
324
361
        stream = StringIO("content")
325
362
        my_store.add(stream, "foo")
326
363
        self.assertEqual(set(['foo']),
335
372
    def test___iter__compressed(self):
336
373
        self.assertEqual(set(['foo']),
337
374
                         set(self.get_populated_store(
338
 
                             store_class=CompressedTextStore).__iter__()))
 
375
                             compressed=True).__iter__()))
339
376
        self.assertEqual(set(['foo']),
340
377
                         set(self.get_populated_store(
341
 
                             True, CompressedTextStore).__iter__()))
 
378
                             True, compressed=True).__iter__()))
342
379
 
343
380
    def test___len__(self):
344
381
        self.assertEqual(1, len(self.get_populated_store()))
345
382
 
346
383
    def test_copy_suffixes(self):
347
384
        from_store = self.get_populated_store()
348
 
        to_store = CompressedTextStore(MemoryTransport(), True)
 
385
        to_store = TextStore(MemoryTransport(), True, compressed=True)
349
386
        to_store.register_suffix('sig')
350
387
        copy_all(from_store, to_store)
351
388
        self.assertEqual(1, len(to_store))
353
390
        self.assertEqual('content', to_store.get('foo').read())
354
391
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
355
392
        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('%'))