← 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: Ian Clatworthy
  • Date: 2007-08-13 14:33:10 UTC
  • mto: (2733.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 2734.
  • Revision ID: ian.clatworthy@internode.on.net-20070813143310-twhj4la0qnupvze8
Added Quick Start Summary

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
1
 
# Copyright (C) 2005-2009, 2011 Canonical Ltd
 
1
# Copyright (C) 2005 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Test Store implementations."""
18
18
 
21
21
import gzip
22
22
 
23
23
import bzrlib.errors as errors
24
 
from bzrlib.errors import BzrError
25
 
from bzrlib.store import TransportStore
 
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
 
25
from bzrlib.transport.local import LocalTransport
26
26
from bzrlib.store.text import TextStore
27
 
from bzrlib.store.versioned import VersionedFileStore
28
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
 
28
import bzrlib.store as store
29
29
import bzrlib.transactions as transactions
30
30
import bzrlib.transport as transport
31
31
from bzrlib.transport.memory import MemoryTransport
32
 
from bzrlib.weave import WeaveFile
33
32
 
34
33
 
35
34
class TestStores(object):
39
38
        f = store.get(fileid)
40
39
        self.assertEqual(f.read(), value)
41
40
 
 
41
    def test_add_str_deprecated(self):
 
42
        os.mkdir('a')
 
43
        store = self.get_store('a')
 
44
        self.callDeprecated(['Passing a string to Store.add'
 
45
                             ' was deprecated in version 0.11.'],
 
46
                            store.add, 'foo', '1')
 
47
        self.assertEqual('foo', store.get('1').read())
 
48
 
42
49
    def fill_store(self, store):
43
50
        store.add(StringIO('hello'), 'a')
44
51
        store.add(StringIO('other'), 'b')
64
71
    def test_get(self):
65
72
        store = self.get_store()
66
73
        self.fill_store(store)
67
 
 
 
74
    
68
75
        self.check_content(store, 'a', 'hello')
69
76
        self.check_content(store, 'b', 'other')
70
77
        self.check_content(store, 'c', 'something')
71
 
 
 
78
    
72
79
        # Make sure that requesting a non-existing file fails
73
80
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
74
81
 
82
89
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
83
90
 
84
91
    def get_store(self, path=u'.'):
85
 
        t = transport.get_transport_from_path(path)
 
92
        t = transport.get_transport(path)
86
93
        return TextStore(t, compressed=True)
87
94
 
88
95
    def test_total_size(self):
92
99
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
100
        # these get gzipped - content should be stable
94
101
        self.assertEqual(store.total_size(), (2, 55))
95
 
 
 
102
        
96
103
    def test__relpath_suffixed(self):
97
104
        my_store = TextStore(MockTransport(),
98
105
                             prefixed=True, compressed=True)
101
108
 
102
109
 
103
110
class TestMemoryStore(TestCase):
104
 
 
 
111
    
105
112
    def get_store(self):
106
113
        return TextStore(MemoryTransport())
107
 
 
 
114
    
108
115
    def test_add_and_retrieve(self):
109
116
        store = self.get_store()
110
117
        store.add(StringIO('hello'), 'aa')
116
123
 
117
124
    def test_missing_is_absent(self):
118
125
        store = self.get_store()
119
 
        self.assertFalse('aa' in store)
 
126
        self.failIf('aa' in store)
120
127
 
121
128
    def test_adding_fails_when_present(self):
122
129
        my_store = self.get_store()
139
146
class TestTextStore(TestCaseInTempDir, TestStores):
140
147
 
141
148
    def get_store(self, path=u'.'):
142
 
        t = transport.get_transport_from_path(path)
 
149
        t = transport.get_transport(path)
143
150
        return TextStore(t, compressed=False)
144
151
 
145
152
    def test_total_size(self):
157
164
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
165
 
159
166
    def get_store(self, path=u'.', compressed=True):
160
 
        t = transport.get_transport_from_path(path)
 
167
        t = transport.get_transport(path)
161
168
        return TextStore(t, compressed=compressed)
162
169
 
163
170
    def test_get_mixed(self):
165
172
        s = self.get_store(u'.', compressed=False)
166
173
        cs.add(StringIO('hello there'), 'a')
167
174
 
168
 
        self.assertPathExists('a.gz')
169
 
        self.assertFalse(os.path.lexists('a'))
 
175
        self.failUnlessExists('a.gz')
 
176
        self.failIf(os.path.lexists('a'))
170
177
 
171
178
        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
172
179
 
174
181
        self.assertEquals(s.has_id('a'), True)
175
182
        self.assertEquals(cs.get('a').read(), 'hello there')
176
183
        self.assertEquals(s.get('a').read(), 'hello there')
177
 
 
 
184
        
178
185
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
179
186
 
180
187
        s.add(StringIO('goodbye'), 'b')
181
 
        self.assertPathExists('b')
182
 
        self.assertFalse(os.path.lexists('b.gz'))
 
188
        self.failUnlessExists('b')
 
189
        self.failIf(os.path.lexists('b.gz'))
183
190
        self.assertEquals(open('b').read(), 'goodbye')
184
191
 
185
192
        self.assertEquals(cs.has_id('b'), True)
186
193
        self.assertEquals(s.has_id('b'), True)
187
194
        self.assertEquals(cs.get('b').read(), 'goodbye')
188
195
        self.assertEquals(s.get('b').read(), 'goodbye')
189
 
 
 
196
        
190
197
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
191
198
 
192
199
class MockTransport(transport.Transport):
204
211
        return
205
212
 
206
213
 
207
 
class InstrumentedTransportStore(TransportStore):
 
214
class InstrumentedTransportStore(store.TransportStore):
208
215
    """An instrumented TransportStore.
209
216
 
210
217
    Here we replace template method worker methods with calls that record the
230
237
class TestMockTransport(TestCase):
231
238
 
232
239
    def test_isinstance(self):
233
 
        self.assertIsInstance(MockTransport(), transport.Transport)
 
240
        self.failUnless(isinstance(MockTransport(), transport.Transport))
234
241
 
235
242
    def test_has(self):
236
243
        self.assertEqual(False, MockTransport().has('foo'))
240
247
 
241
248
 
242
249
class TestTransportStore(TestCase):
243
 
 
 
250
    
244
251
    def test__relpath_invalid(self):
245
 
        my_store = TransportStore(MockTransport())
 
252
        my_store = store.TransportStore(MockTransport())
246
253
        self.assertRaises(ValueError, my_store._relpath, '/foo')
247
254
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
248
255
 
249
256
    def test_register_invalid_suffixes(self):
250
 
        my_store = TransportStore(MockTransport())
 
257
        my_store = store.TransportStore(MockTransport())
251
258
        self.assertRaises(ValueError, my_store.register_suffix, '/')
252
259
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
253
260
 
254
261
    def test__relpath_unregister_suffixes(self):
255
 
        my_store = TransportStore(MockTransport())
 
262
        my_store = store.TransportStore(MockTransport())
256
263
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
257
264
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
258
265
 
259
266
    def test__relpath_simple(self):
260
 
        my_store = TransportStore(MockTransport())
 
267
        my_store = store.TransportStore(MockTransport())
261
268
        self.assertEqual("foo", my_store._relpath('foo'))
262
269
 
263
270
    def test__relpath_prefixed(self):
264
 
        my_store = TransportStore(MockTransport(), True)
 
271
        my_store = store.TransportStore(MockTransport(), True)
265
272
        self.assertEqual('45/foo', my_store._relpath('foo'))
266
273
 
267
274
    def test__relpath_simple_suffixed(self):
268
 
        my_store = TransportStore(MockTransport())
 
275
        my_store = store.TransportStore(MockTransport())
269
276
        my_store.register_suffix('bar')
270
277
        my_store.register_suffix('baz')
271
278
        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
272
279
        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
273
280
 
274
281
    def test__relpath_prefixed_suffixed(self):
275
 
        my_store = TransportStore(MockTransport(), True)
 
282
        my_store = store.TransportStore(MockTransport(), True)
276
283
        my_store.register_suffix('bar')
277
284
        my_store.register_suffix('baz')
278
285
        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
297
304
        my_store.register_suffix('dsc')
298
305
        my_store.add(stream, "foo", 'dsc')
299
306
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
300
 
 
 
307
        
301
308
    def test_add_simple_suffixed(self):
302
309
        stream = StringIO("content")
303
310
        my_store = InstrumentedTransportStore(MockTransport(), True)
317
324
        stream = StringIO("signature for missing base")
318
325
        my_store.add(stream, "missing", 'sig')
319
326
        return my_store
320
 
 
 
327
        
321
328
    def test_has_simple(self):
322
329
        my_store = self.get_populated_store()
323
330
        self.assertEqual(True, my_store.has_id('foo'))
394
401
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
395
402
 
396
403
    def test_relpath_escaped(self):
397
 
        my_store = TransportStore(MemoryTransport())
 
404
        my_store = store.TransportStore(MemoryTransport())
398
405
        self.assertEqual('%25', my_store._relpath('%'))
399
406
 
400
407
    def test_escaped_uppercase(self):
401
408
        """Uppercase letters are escaped for safety on Windows"""
402
 
        my_store = TransportStore(MemoryTransport(), prefixed=True,
403
 
            escaped=True)
 
409
        my_store = store.TransportStore(MemoryTransport(), escaped=True)
404
410
        # a particularly perverse file-id! :-)
405
 
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
 
411
        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
406
412
 
407
413
 
408
414
class TestVersionFileStore(TestCaseWithTransport):
409
415
 
410
 
    def get_scope(self):
411
 
        return self._transaction
412
 
 
413
416
    def setUp(self):
414
417
        super(TestVersionFileStore, self).setUp()
415
 
        self.vfstore = VersionedFileStore(MemoryTransport(),
416
 
            versionedfile_class=WeaveFile)
417
 
        self.vfstore.get_scope = self.get_scope
418
 
        self._transaction = None
 
418
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
419
419
 
420
420
    def test_get_weave_registers_dirty_in_write(self):
421
 
        self._transaction = transactions.WriteTransaction()
422
 
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
423
 
        self._transaction.finish()
424
 
        self._transaction = None
425
 
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
426
 
        self._transaction = transactions.WriteTransaction()
427
 
        vf = self.vfstore.get_weave('id', self._transaction)
428
 
        self._transaction.finish()
429
 
        self._transaction = None
430
 
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
421
        transaction = transactions.WriteTransaction()
 
422
        vf = self.vfstore.get_weave_or_empty('id', transaction)
 
423
        transaction.finish()
 
424
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
425
        transaction = transactions.WriteTransaction()
 
426
        vf = self.vfstore.get_weave('id', transaction)
 
427
        transaction.finish()
 
428
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
429
 
 
430
    def test_get_weave_or_empty_readonly_fails(self):
 
431
        transaction = transactions.ReadOnlyTransaction()
 
432
        vf = self.assertRaises(errors.ReadOnlyError,
 
433
                               self.vfstore.get_weave_or_empty,
 
434
                               'id',
 
435
                               transaction)
431
436
 
432
437
    def test_get_weave_readonly_cant_write(self):
433
 
        self._transaction = transactions.WriteTransaction()
434
 
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
435
 
        self._transaction.finish()
436
 
        self._transaction = transactions.ReadOnlyTransaction()
437
 
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
438
        transaction = transactions.WriteTransaction()
 
439
        vf = self.vfstore.get_weave_or_empty('id', transaction)
 
440
        transaction.finish()
 
441
        transaction = transactions.ReadOnlyTransaction()
 
442
        vf = self.vfstore.get_weave_or_empty('id', transaction)
438
443
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
439
444
 
440
 
    def test___iter__escaped(self):
441
 
        self.vfstore = VersionedFileStore(MemoryTransport(),
442
 
            prefixed=True, escaped=True, versionedfile_class=WeaveFile)
443
 
        self.vfstore.get_scope = self.get_scope
444
 
        self._transaction = transactions.WriteTransaction()
445
 
        vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
446
 
        vf.add_lines('a', [], [])
447
 
        del vf
448
 
        self._transaction.finish()
449
 
        self.assertEqual([' '], list(self.vfstore))