/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Martin
  • Date: 2017-05-21 18:16:32 UTC
  • mto: (6621.2.22 py3)
  • mto: This revision was merged to the branch mainline in revision 6624.
  • Revision ID: gzlist@googlemail.com-20170521181632-kll0wqnsq8pipfpj
Use BytesIO or StringIO from bzrlib.sixish

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
16
16
 
17
17
"""Test Store implementations."""
18
18
 
19
 
from cStringIO import StringIO
20
19
import os
21
20
import gzip
22
21
 
23
22
import bzrlib.errors as errors
24
23
from bzrlib.errors import BzrError
 
24
from bzrlib.sixish import (
 
25
    BytesIO,
 
26
    )
25
27
from bzrlib.store import TransportStore
26
28
from bzrlib.store.text import TextStore
27
29
from bzrlib.store.versioned import VersionedFileStore
40
42
        self.assertEqual(f.read(), value)
41
43
 
42
44
    def fill_store(self, store):
43
 
        store.add(StringIO('hello'), 'a')
44
 
        store.add(StringIO('other'), 'b')
45
 
        store.add(StringIO('something'), 'c')
46
 
        store.add(StringIO('goodbye'), '123123')
 
45
        store.add(BytesIO(b'hello'), 'a')
 
46
        store.add(BytesIO(b'other'), 'b')
 
47
        store.add(BytesIO(b'something'), 'c')
 
48
        store.add(BytesIO(b'goodbye'), '123123')
47
49
 
48
50
    def test_copy_all(self):
49
51
        """Test copying"""
50
52
        os.mkdir('a')
51
53
        store_a = self.get_store('a')
52
 
        store_a.add(StringIO('foo'), '1')
 
54
        store_a.add(BytesIO(b'foo'), '1')
53
55
        os.mkdir('b')
54
56
        store_b = self.get_store('b')
55
57
        store_b.copy_all_ids(store_a)
76
78
        """Multiple add with same ID should raise a BzrError"""
77
79
        store = self.get_store()
78
80
        self.fill_store(store)
79
 
        self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
81
        self.assertRaises(BzrError, store.add, BytesIO(b'goodbye'), '123123')
80
82
 
81
83
 
82
84
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
88
90
    def test_total_size(self):
89
91
        store = self.get_store(u'.')
90
92
        store.register_suffix('dsc')
91
 
        store.add(StringIO('goodbye'), '123123')
92
 
        store.add(StringIO('goodbye2'), '123123', 'dsc')
 
93
        store.add(BytesIO(b'goodbye'), '123123')
 
94
        store.add(BytesIO(b'goodbye2'), '123123', 'dsc')
93
95
        # these get gzipped - content should be stable
94
96
        self.assertEqual(store.total_size(), (2, 55))
95
97
 
107
109
 
108
110
    def test_add_and_retrieve(self):
109
111
        store = self.get_store()
110
 
        store.add(StringIO('hello'), 'aa')
 
112
        store.add(BytesIO(b'hello'), 'aa')
111
113
        self.assertNotEqual(store.get('aa'), None)
112
114
        self.assertEqual(store.get('aa').read(), 'hello')
113
 
        store.add(StringIO('hello world'), 'bb')
 
115
        store.add(BytesIO(b'hello world'), 'bb')
114
116
        self.assertNotEqual(store.get('bb'), None)
115
117
        self.assertEqual(store.get('bb').read(), 'hello world')
116
118
 
120
122
 
121
123
    def test_adding_fails_when_present(self):
122
124
        my_store = self.get_store()
123
 
        my_store.add(StringIO('hello'), 'aa')
 
125
        my_store.add(BytesIO(b'hello'), 'aa')
124
126
        self.assertRaises(BzrError,
125
 
                          my_store.add, StringIO('hello'), 'aa')
 
127
                          my_store.add, BytesIO(b'hello'), 'aa')
126
128
 
127
129
    def test_total_size(self):
128
130
        store = self.get_store()
129
 
        store.add(StringIO('goodbye'), '123123')
130
 
        store.add(StringIO('goodbye2'), '123123.dsc')
 
131
        store.add(BytesIO(b'goodbye'), '123123')
 
132
        store.add(BytesIO(b'goodbye2'), '123123.dsc')
131
133
        self.assertEqual(store.total_size(), (2, 15))
132
134
        # TODO: Switch the exception form UnlistableStore to
133
135
        #       or make Stores throw UnlistableStore if their
144
146
 
145
147
    def test_total_size(self):
146
148
        store = self.get_store()
147
 
        store.add(StringIO('goodbye'), '123123')
148
 
        store.add(StringIO('goodbye2'), '123123.dsc')
 
149
        store.add(BytesIO(b'goodbye'), '123123')
 
150
        store.add(BytesIO(b'goodbye2'), '123123.dsc')
149
151
        self.assertEqual(store.total_size(), (2, 15))
150
152
        # TODO: Switch the exception form UnlistableStore to
151
153
        #       or make Stores throw UnlistableStore if their
163
165
    def test_get_mixed(self):
164
166
        cs = self.get_store(u'.', compressed=True)
165
167
        s = self.get_store(u'.', compressed=False)
166
 
        cs.add(StringIO('hello there'), 'a')
 
168
        cs.add(BytesIO(b'hello there'), 'a')
167
169
 
168
170
        self.assertPathExists('a.gz')
169
171
        self.assertFalse(os.path.lexists('a'))
175
177
        self.assertEqual(cs.get('a').read(), 'hello there')
176
178
        self.assertEqual(s.get('a').read(), 'hello there')
177
179
 
178
 
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
 
180
        self.assertRaises(BzrError, s.add, BytesIO(b'goodbye'), 'a')
179
181
 
180
 
        s.add(StringIO('goodbye'), 'b')
 
182
        s.add(BytesIO(b'goodbye'), 'b')
181
183
        self.assertPathExists('b')
182
184
        self.assertFalse(os.path.lexists('b.gz'))
183
185
        self.assertEqual(open('b').read(), 'goodbye')
187
189
        self.assertEqual(cs.get('b').read(), 'goodbye')
188
190
        self.assertEqual(s.get('b').read(), 'goodbye')
189
191
 
190
 
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
 
192
        self.assertRaises(BzrError, cs.add, BytesIO(b'again'), 'b')
191
193
 
192
194
class MockTransport(transport.Transport):
193
195
    """A fake transport for testing with."""
280
282
                         my_store._relpath('foo', ['bar', 'baz']))
281
283
 
282
284
    def test_add_simple(self):
283
 
        stream = StringIO("content")
 
285
        stream = BytesIO(b"content")
284
286
        my_store = InstrumentedTransportStore(MockTransport())
285
287
        my_store.add(stream, "foo")
286
288
        self.assertEqual([("_add", "foo", stream)], my_store._calls)
287
289
 
288
290
    def test_add_prefixed(self):
289
 
        stream = StringIO("content")
 
291
        stream = BytesIO(b"content")
290
292
        my_store = InstrumentedTransportStore(MockTransport(), True)
291
293
        my_store.add(stream, "foo")
292
294
        self.assertEqual([("_add", "45/foo", stream)], my_store._calls)
293
295
 
294
296
    def test_add_simple_suffixed(self):
295
 
        stream = StringIO("content")
 
297
        stream = BytesIO(b"content")
296
298
        my_store = InstrumentedTransportStore(MockTransport())
297
299
        my_store.register_suffix('dsc')
298
300
        my_store.add(stream, "foo", 'dsc')
299
301
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
300
302
 
301
303
    def test_add_simple_suffixed(self):
302
 
        stream = StringIO("content")
 
304
        stream = BytesIO(b"content")
303
305
        my_store = InstrumentedTransportStore(MockTransport(), True)
304
306
        my_store.register_suffix('dsc')
305
307
        my_store.add(stream, "foo", 'dsc')
310
312
        my_store = store_class(MemoryTransport(), prefixed,
311
313
                               compressed=compressed)
312
314
        my_store.register_suffix('sig')
313
 
        stream = StringIO("signature")
 
315
        stream = BytesIO(b"signature")
314
316
        my_store.add(stream, "foo", 'sig')
315
 
        stream = StringIO("content")
 
317
        stream = BytesIO(b"content")
316
318
        my_store.add(stream, "foo")
317
 
        stream = StringIO("signature for missing base")
 
319
        stream = BytesIO(b"signature for missing base")
318
320
        my_store.add(stream, "missing", 'sig')
319
321
        return my_store
320
322
 
359
361
    def test___iter__no_suffix(self):
360
362
        my_store = TextStore(MemoryTransport(),
361
363
                             prefixed=False, compressed=False)
362
 
        stream = StringIO("content")
 
364
        stream = BytesIO(b"content")
363
365
        my_store.add(stream, "foo")
364
366
        self.assertEqual({'foo'},
365
367
                         set(my_store.__iter__()))