~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

Disable selftest that was creating .bazaar in source directory

Show diffs side-by-side

added added

removed removed

Lines of Context:
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(), True, compressed=True)
 
96
        my_store = CompressedTextStore(MockTransport(), True)
96
97
        my_store.register_suffix('dsc')
97
 
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
98
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
98
99
 
99
100
 
100
101
class TestMemoryStore(TestCase):
138
139
 
139
140
class TestTextStore(TestCaseInTempDir, TestStores):
140
141
 
141
 
    def get_store(self, path=u'.'):
 
142
    def get_store(self, path='.'):
142
143
        t = LocalTransport(path)
143
 
        return TextStore(t, compressed=False)
 
144
        return TextStore(t)
144
145
 
145
146
    def test_total_size(self):
146
147
        store = self.get_store()
154
155
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
155
156
 
156
157
 
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
158
class MockTransport(transport.Transport):
193
159
    """A fake transport for testing with."""
194
160
 
266
232
 
267
233
    def test__relpath_simple_suffixed(self):
268
234
        my_store = store.TransportStore(MockTransport())
 
235
        my_store.register_suffix('gz')
269
236
        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']))
 
237
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
238
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
273
239
 
274
240
    def test__relpath_prefixed_suffixed(self):
275
241
        my_store = store.TransportStore(MockTransport(), True)
 
242
        my_store.register_suffix('gz')
276
243
        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']))
 
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']))
281
247
 
282
248
    def test_add_simple(self):
283
249
        stream = StringIO("content")
305
271
        my_store.add(stream, "foo", 'dsc')
306
272
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
307
273
 
308
 
    def get_populated_store(self, prefixed=False,
309
 
            store_class=TextStore, compressed=False):
310
 
        my_store = store_class(MemoryTransport(), prefixed,
311
 
                               compressed=compressed)
 
274
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
275
        my_store = store_class(MemoryTransport(), prefixed)
312
276
        my_store.register_suffix('sig')
313
277
        stream = StringIO("signature")
314
278
        my_store.add(stream, "foo", 'sig')
357
321
                         my_store.get('missing', 'sig').read())
358
322
 
359
323
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
324
        my_store = TextStore(MemoryTransport(), False)
361
325
        stream = StringIO("content")
362
326
        my_store.add(stream, "foo")
363
327
        self.assertEqual(set(['foo']),
372
336
    def test___iter__compressed(self):
373
337
        self.assertEqual(set(['foo']),
374
338
                         set(self.get_populated_store(
375
 
                             compressed=True).__iter__()))
 
339
                             store_class=CompressedTextStore).__iter__()))
376
340
        self.assertEqual(set(['foo']),
377
341
                         set(self.get_populated_store(
378
 
                             True, compressed=True).__iter__()))
 
342
                             True, CompressedTextStore).__iter__()))
379
343
 
380
344
    def test___len__(self):
381
345
        self.assertEqual(1, len(self.get_populated_store()))
382
346
 
383
347
    def test_copy_suffixes(self):
384
348
        from_store = self.get_populated_store()
385
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
349
        to_store = CompressedTextStore(MemoryTransport(), True)
386
350
        to_store.register_suffix('sig')
387
351
        copy_all(from_store, to_store)
388
352
        self.assertEqual(1, len(to_store))