/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 breezy/tests/test_store.py

  • Committer: Martin
  • Date: 2017-06-05 20:48:31 UTC
  • mto: This revision was merged to the branch mainline in revision 6658.
  • Revision ID: gzlist@googlemail.com-20170605204831-20accykspjcrx0a8
Apply 2to3 dict fixer and clean up resulting mess using view helpers

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_store.py
1
 
# Copyright (C) 2005, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2011, 2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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
 
import bzrlib.errors as errors
24
 
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
25
 
from bzrlib.transport.local import LocalTransport
26
 
from bzrlib.store.text import TextStore
27
 
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
 
import bzrlib.store as store
29
 
import bzrlib.store.versioned
30
 
import bzrlib.transactions as transactions
31
 
import bzrlib.transport as transport
32
 
from bzrlib.transport.memory import MemoryTransport
 
22
from .. import errors as errors
 
23
from ..errors import BzrError
 
24
from ..sixish import (
 
25
    BytesIO,
 
26
    )
 
27
from ..store import TransportStore
 
28
from ..store.text import TextStore
 
29
from ..store.versioned import VersionedFileStore
 
30
from . import TestCase, TestCaseInTempDir, TestCaseWithTransport
 
31
from .. import transactions as transactions
 
32
from .. import transport as transport
 
33
from ..transport.memory import MemoryTransport
 
34
from ..weave import WeaveFile
33
35
 
34
36
 
35
37
class TestStores(object):
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):
83
85
 
84
86
    def get_store(self, path=u'.'):
85
 
        t = transport.get_transport(path)
 
87
        t = transport.get_transport_from_path(path)
86
88
        return TextStore(t, compressed=True)
87
89
 
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
 
117
119
    def test_missing_is_absent(self):
118
120
        store = self.get_store()
119
 
        self.failIf('aa' in store)
 
121
        self.assertFalse('aa' in store)
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
139
141
class TestTextStore(TestCaseInTempDir, TestStores):
140
142
 
141
143
    def get_store(self, path=u'.'):
142
 
        t = transport.get_transport(path)
 
144
        t = transport.get_transport_from_path(path)
143
145
        return TextStore(t, compressed=False)
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
157
159
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
160
 
159
161
    def get_store(self, path=u'.', compressed=True):
160
 
        t = transport.get_transport(path)
 
162
        t = transport.get_transport_from_path(path)
161
163
        return TextStore(t, compressed=compressed)
162
164
 
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')
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')
 
168
        cs.add(BytesIO(b'hello there'), 'a')
 
169
 
 
170
        self.assertPathExists('a.gz')
 
171
        self.assertFalse(os.path.lexists('a'))
 
172
 
 
173
        self.assertEqual(gzip.GzipFile('a.gz').read(), 'hello there')
 
174
 
 
175
        self.assertEqual(cs.has_id('a'), True)
 
176
        self.assertEqual(s.has_id('a'), True)
 
177
        self.assertEqual(cs.get('a').read(), 'hello there')
 
178
        self.assertEqual(s.get('a').read(), 'hello there')
 
179
 
 
180
        self.assertRaises(BzrError, s.add, BytesIO(b'goodbye'), 'a')
 
181
 
 
182
        s.add(BytesIO(b'goodbye'), 'b')
 
183
        self.assertPathExists('b')
 
184
        self.assertFalse(os.path.lexists('b.gz'))
 
185
        self.assertEqual(open('b').read(), 'goodbye')
 
186
 
 
187
        self.assertEqual(cs.has_id('b'), True)
 
188
        self.assertEqual(s.has_id('b'), True)
 
189
        self.assertEqual(cs.get('b').read(), 'goodbye')
 
190
        self.assertEqual(s.get('b').read(), 'goodbye')
 
191
 
 
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."""
204
206
        return
205
207
 
206
208
 
207
 
class InstrumentedTransportStore(store.TransportStore):
 
209
class InstrumentedTransportStore(TransportStore):
208
210
    """An instrumented TransportStore.
209
211
 
210
212
    Here we replace template method worker methods with calls that record the
230
232
class TestMockTransport(TestCase):
231
233
 
232
234
    def test_isinstance(self):
233
 
        self.failUnless(isinstance(MockTransport(), transport.Transport))
 
235
        self.assertIsInstance(MockTransport(), transport.Transport)
234
236
 
235
237
    def test_has(self):
236
238
        self.assertEqual(False, MockTransport().has('foo'))
242
244
class TestTransportStore(TestCase):
243
245
 
244
246
    def test__relpath_invalid(self):
245
 
        my_store = store.TransportStore(MockTransport())
 
247
        my_store = TransportStore(MockTransport())
246
248
        self.assertRaises(ValueError, my_store._relpath, '/foo')
247
249
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
248
250
 
249
251
    def test_register_invalid_suffixes(self):
250
 
        my_store = store.TransportStore(MockTransport())
 
252
        my_store = TransportStore(MockTransport())
251
253
        self.assertRaises(ValueError, my_store.register_suffix, '/')
252
254
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
253
255
 
254
256
    def test__relpath_unregister_suffixes(self):
255
 
        my_store = store.TransportStore(MockTransport())
 
257
        my_store = TransportStore(MockTransport())
256
258
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
257
259
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
258
260
 
259
261
    def test__relpath_simple(self):
260
 
        my_store = store.TransportStore(MockTransport())
 
262
        my_store = TransportStore(MockTransport())
261
263
        self.assertEqual("foo", my_store._relpath('foo'))
262
264
 
263
265
    def test__relpath_prefixed(self):
264
 
        my_store = store.TransportStore(MockTransport(), True)
 
266
        my_store = TransportStore(MockTransport(), True)
265
267
        self.assertEqual('45/foo', my_store._relpath('foo'))
266
268
 
267
269
    def test__relpath_simple_suffixed(self):
268
 
        my_store = store.TransportStore(MockTransport())
 
270
        my_store = TransportStore(MockTransport())
269
271
        my_store.register_suffix('bar')
270
272
        my_store.register_suffix('baz')
271
273
        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
272
274
        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
273
275
 
274
276
    def test__relpath_prefixed_suffixed(self):
275
 
        my_store = store.TransportStore(MockTransport(), True)
 
277
        my_store = TransportStore(MockTransport(), True)
276
278
        my_store.register_suffix('bar')
277
279
        my_store.register_suffix('baz')
278
280
        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
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
 
        self.assertEqual(set(['foo']),
 
366
        self.assertEqual({'foo'},
365
367
                         set(my_store.__iter__()))
366
368
 
367
369
    def test___iter__(self):
368
 
        self.assertEqual(set(['foo']),
 
370
        self.assertEqual({'foo'},
369
371
                         set(self.get_populated_store().__iter__()))
370
 
        self.assertEqual(set(['foo']),
 
372
        self.assertEqual({'foo'},
371
373
                         set(self.get_populated_store(True).__iter__()))
372
374
 
373
375
    def test___iter__compressed(self):
374
 
        self.assertEqual(set(['foo']),
 
376
        self.assertEqual({'foo'},
375
377
                         set(self.get_populated_store(
376
378
                             compressed=True).__iter__()))
377
 
        self.assertEqual(set(['foo']),
 
379
        self.assertEqual({'foo'},
378
380
                         set(self.get_populated_store(
379
381
                             True, compressed=True).__iter__()))
380
382
 
388
390
        to_store.register_suffix('sig')
389
391
        to_store.copy_all_ids(from_store)
390
392
        self.assertEqual(1, len(to_store))
391
 
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
 
393
        self.assertEqual({'foo'}, set(to_store.__iter__()))
392
394
        self.assertEqual('content', to_store.get('foo').read())
393
395
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
394
396
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
395
397
 
396
398
    def test_relpath_escaped(self):
397
 
        my_store = store.TransportStore(MemoryTransport())
 
399
        my_store = TransportStore(MemoryTransport())
398
400
        self.assertEqual('%25', my_store._relpath('%'))
399
401
 
400
402
    def test_escaped_uppercase(self):
401
403
        """Uppercase letters are escaped for safety on Windows"""
402
 
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
 
404
        my_store = TransportStore(MemoryTransport(), prefixed=True,
403
405
            escaped=True)
404
406
        # a particularly perverse file-id! :-)
405
 
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
 
407
        self.assertEqual(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
406
408
 
407
409
 
408
410
class TestVersionFileStore(TestCaseWithTransport):
412
414
 
413
415
    def setUp(self):
414
416
        super(TestVersionFileStore, self).setUp()
415
 
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
417
        self.vfstore = VersionedFileStore(MemoryTransport(),
 
418
            versionedfile_class=WeaveFile)
416
419
        self.vfstore.get_scope = self.get_scope
417
420
        self._transaction = None
418
421
 
437
440
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
438
441
 
439
442
    def test___iter__escaped(self):
440
 
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
441
 
            prefixed=True, escaped=True)
 
443
        self.vfstore = VersionedFileStore(MemoryTransport(),
 
444
            prefixed=True, escaped=True, versionedfile_class=WeaveFile)
442
445
        self.vfstore.get_scope = self.get_scope
443
446
        self._transaction = transactions.WriteTransaction()
444
447
        vf = self.vfstore.get_weave_or_empty(' ', self._transaction)