← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

Fix http server tests for win32/macos (John)

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
22
from bzrlib.errors import BzrError, UnlistableStore
24
23
from bzrlib.store import copy_all
25
24
from bzrlib.transport.local import LocalTransport
26
25
from bzrlib.transport import NoSuchFile
 
26
from bzrlib.store.compressed_text import CompressedTextStore
27
27
from bzrlib.store.text import TextStore
28
 
from bzrlib.tests import TestCase, TestCaseInTempDir
 
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
29
29
import bzrlib.store as store
30
30
import bzrlib.transport as transport
31
31
from bzrlib.transport.memory import MemoryTransport
32
32
 
33
33
 
34
34
class TestStores(object):
35
 
    """Mixin template class that provides some common tests for stores"""
36
35
 
37
36
    def check_content(self, store, fileid, value):
38
37
        f = store.get(fileid)
80
79
 
81
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
81
 
83
 
    def get_store(self, path=u'.'):
 
82
    def get_store(self, path='.'):
84
83
        t = LocalTransport(path)
85
 
        return TextStore(t, compressed=True)
 
84
        return CompressedTextStore(t)
86
85
 
87
86
    def test_total_size(self):
88
 
        store = self.get_store(u'.')
 
87
        store = self.get_store('.')
89
88
        store.register_suffix('dsc')
90
89
        store.add(StringIO('goodbye'), '123123')
91
90
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
92
        self.assertEqual(store.total_size(), (2, 55))
94
93
        
95
94
    def test__relpath_suffixed(self):
96
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
95
        my_store = CompressedTextStore(MockTransport(), True)
97
96
        my_store.register_suffix('dsc')
98
 
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
97
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
99
98
 
100
99
 
101
100
class TestMemoryStore(TestCase):
139
138
 
140
139
class TestTextStore(TestCaseInTempDir, TestStores):
141
140
 
142
 
    def get_store(self, path=u'.'):
 
141
    def get_store(self, path='.'):
143
142
        t = LocalTransport(path)
144
 
        return TextStore(t, compressed=False)
 
143
        return TextStore(t)
145
144
 
146
145
    def test_total_size(self):
147
146
        store = self.get_store()
155
154
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
156
155
 
157
156
 
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
157
class MockTransport(transport.Transport):
194
158
    """A fake transport for testing with."""
195
159
 
267
231
 
268
232
    def test__relpath_simple_suffixed(self):
269
233
        my_store = store.TransportStore(MockTransport())
 
234
        my_store.register_suffix('gz')
270
235
        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']))
 
236
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
237
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
274
238
 
275
239
    def test__relpath_prefixed_suffixed(self):
276
240
        my_store = store.TransportStore(MockTransport(), True)
 
241
        my_store.register_suffix('gz')
277
242
        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']))
 
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']))
282
246
 
283
247
    def test_add_simple(self):
284
248
        stream = StringIO("content")
306
270
        my_store.add(stream, "foo", 'dsc')
307
271
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
308
272
 
309
 
    def get_populated_store(self, prefixed=False,
310
 
            store_class=TextStore, compressed=False):
311
 
        my_store = store_class(MemoryTransport(), prefixed,
312
 
                               compressed=compressed)
 
273
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
274
        my_store = store_class(MemoryTransport(), prefixed)
313
275
        my_store.register_suffix('sig')
314
276
        stream = StringIO("signature")
315
277
        my_store.add(stream, "foo", 'sig')
358
320
                         my_store.get('missing', 'sig').read())
359
321
 
360
322
    def test___iter__no_suffix(self):
361
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
323
        my_store = TextStore(MemoryTransport(), False)
362
324
        stream = StringIO("content")
363
325
        my_store.add(stream, "foo")
364
326
        self.assertEqual(set(['foo']),
373
335
    def test___iter__compressed(self):
374
336
        self.assertEqual(set(['foo']),
375
337
                         set(self.get_populated_store(
376
 
                             compressed=True).__iter__()))
 
338
                             store_class=CompressedTextStore).__iter__()))
377
339
        self.assertEqual(set(['foo']),
378
340
                         set(self.get_populated_store(
379
 
                             True, compressed=True).__iter__()))
 
341
                             True, CompressedTextStore).__iter__()))
380
342
 
381
343
    def test___len__(self):
382
344
        self.assertEqual(1, len(self.get_populated_store()))
383
345
 
384
346
    def test_copy_suffixes(self):
385
347
        from_store = self.get_populated_store()
386
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
348
        to_store = CompressedTextStore(MemoryTransport(), True)
387
349
        to_store.register_suffix('sig')
388
350
        copy_all(from_store, to_store)
389
351
        self.assertEqual(1, len(to_store))