← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: Robert Collins
  • Date: 2005-10-17 11:41:07 UTC
  • mfrom: (1442.1.60)
  • Revision ID: robertc@robertcollins.net-20051017114107-f5586285d825c105
Merge in first part of GPG support.

This adds check_signatures config support, triams back the transport api
to be leaner and easier to hook in suffixes - non primary streams in the store
associated with the fileid that primary data is stored in, a gpg module which
will encapsulate all signing and checking operations.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
1
 
# Copyright (C) 2005-2009, 2011 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2005 by Canonical Development Ltd
 
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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
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
 
19
19
from cStringIO import StringIO
20
20
import os
21
 
import gzip
22
21
 
23
 
import bzrlib.errors as errors
24
 
from bzrlib.errors import BzrError
25
 
from bzrlib.store import TransportStore
 
22
from bzrlib.errors import BzrError, UnlistableStore
 
23
from bzrlib.store import copy_all
 
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.store.versioned import VersionedFileStore
28
 
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
29
 
import bzrlib.transactions as transactions
 
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
29
import bzrlib.store as store
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):
36
 
    """Mixin template class that provides some common tests for stores"""
37
35
 
38
36
    def check_content(self, store, fileid, value):
39
37
        f = store.get(fileid)
49
47
        """Test copying"""
50
48
        os.mkdir('a')
51
49
        store_a = self.get_store('a')
52
 
        store_a.add(StringIO('foo'), '1')
 
50
        store_a.add('foo', '1')
53
51
        os.mkdir('b')
54
52
        store_b = self.get_store('b')
55
 
        store_b.copy_all_ids(store_a)
 
53
        copy_all(store_a, store_b)
56
54
        self.assertEqual(store_a.get('1').read(), 'foo')
57
55
        self.assertEqual(store_b.get('1').read(), 'foo')
58
56
        # TODO: Switch the exception form UnlistableStore to
64
62
    def test_get(self):
65
63
        store = self.get_store()
66
64
        self.fill_store(store)
67
 
 
 
65
    
68
66
        self.check_content(store, 'a', 'hello')
69
67
        self.check_content(store, 'b', 'other')
70
68
        self.check_content(store, 'c', 'something')
71
 
 
 
69
    
72
70
        # Make sure that requesting a non-existing file fails
73
71
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
74
72
 
81
79
 
82
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
83
81
 
84
 
    def get_store(self, path=u'.'):
85
 
        t = transport.get_transport(path)
86
 
        return TextStore(t, compressed=True)
 
82
    def get_store(self, path='.'):
 
83
        t = LocalTransport(path)
 
84
        return CompressedTextStore(t)
87
85
 
88
86
    def test_total_size(self):
89
 
        store = self.get_store(u'.')
 
87
        store = self.get_store('.')
90
88
        store.register_suffix('dsc')
91
89
        store.add(StringIO('goodbye'), '123123')
92
90
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
91
        # these get gzipped - content should be stable
94
92
        self.assertEqual(store.total_size(), (2, 55))
95
 
 
 
93
        
96
94
    def test__relpath_suffixed(self):
97
 
        my_store = TextStore(MockTransport(),
98
 
                             prefixed=True, compressed=True)
 
95
        my_store = CompressedTextStore(MockTransport(), True)
99
96
        my_store.register_suffix('dsc')
100
 
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
97
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
101
98
 
102
99
 
103
100
class TestMemoryStore(TestCase):
104
 
 
 
101
    
105
102
    def get_store(self):
106
 
        return TextStore(MemoryTransport())
 
103
        return store.ImmutableMemoryStore()
 
104
    
 
105
    def test_imports(self):
 
106
        from bzrlib.store import ImmutableMemoryStore
107
107
 
108
108
    def test_add_and_retrieve(self):
109
109
        store = self.get_store()
116
116
 
117
117
    def test_missing_is_absent(self):
118
118
        store = self.get_store()
119
 
        self.assertFalse('aa' in store)
 
119
        self.failIf('aa' in store)
120
120
 
121
121
    def test_adding_fails_when_present(self):
122
122
        my_store = self.get_store()
138
138
 
139
139
class TestTextStore(TestCaseInTempDir, TestStores):
140
140
 
141
 
    def get_store(self, path=u'.'):
142
 
        t = transport.get_transport(path)
143
 
        return TextStore(t, compressed=False)
 
141
    def get_store(self, path='.'):
 
142
        t = LocalTransport(path)
 
143
        return TextStore(t)
144
144
 
145
145
    def test_total_size(self):
146
146
        store = self.get_store()
154
154
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
155
155
 
156
156
 
157
 
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
 
 
159
 
    def get_store(self, path=u'.', compressed=True):
160
 
        t = transport.get_transport(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.assertPathExists('a.gz')
169
 
        self.assertFalse(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.assertPathExists('b')
182
 
        self.assertFalse(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
157
class MockTransport(transport.Transport):
193
158
    """A fake transport for testing with."""
194
159
 
204
169
        return
205
170
 
206
171
 
207
 
class InstrumentedTransportStore(TransportStore):
 
172
class InstrumentedTransportStore(store.TransportStore):
208
173
    """An instrumented TransportStore.
209
174
 
210
175
    Here we replace template method worker methods with calls that record the
230
195
class TestMockTransport(TestCase):
231
196
 
232
197
    def test_isinstance(self):
233
 
        self.assertIsInstance(MockTransport(), transport.Transport)
 
198
        self.failUnless(isinstance(MockTransport(), transport.Transport))
234
199
 
235
200
    def test_has(self):
236
201
        self.assertEqual(False, MockTransport().has('foo'))
240
205
 
241
206
 
242
207
class TestTransportStore(TestCase):
243
 
 
 
208
    
244
209
    def test__relpath_invalid(self):
245
 
        my_store = TransportStore(MockTransport())
 
210
        my_store = store.TransportStore(MockTransport())
246
211
        self.assertRaises(ValueError, my_store._relpath, '/foo')
247
212
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
248
213
 
249
214
    def test_register_invalid_suffixes(self):
250
 
        my_store = TransportStore(MockTransport())
 
215
        my_store = store.TransportStore(MockTransport())
251
216
        self.assertRaises(ValueError, my_store.register_suffix, '/')
252
217
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
253
218
 
254
219
    def test__relpath_unregister_suffixes(self):
255
 
        my_store = TransportStore(MockTransport())
 
220
        my_store = store.TransportStore(MockTransport())
256
221
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
257
222
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
258
223
 
259
224
    def test__relpath_simple(self):
260
 
        my_store = TransportStore(MockTransport())
 
225
        my_store = store.TransportStore(MockTransport())
261
226
        self.assertEqual("foo", my_store._relpath('foo'))
262
227
 
263
228
    def test__relpath_prefixed(self):
264
 
        my_store = TransportStore(MockTransport(), True)
 
229
        my_store = store.TransportStore(MockTransport(), True)
265
230
        self.assertEqual('45/foo', my_store._relpath('foo'))
266
231
 
267
232
    def test__relpath_simple_suffixed(self):
268
 
        my_store = TransportStore(MockTransport())
 
233
        my_store = store.TransportStore(MockTransport())
 
234
        my_store.register_suffix('gz')
269
235
        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']))
 
236
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
237
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
273
238
 
274
239
    def test__relpath_prefixed_suffixed(self):
275
 
        my_store = TransportStore(MockTransport(), True)
 
240
        my_store = store.TransportStore(MockTransport(), True)
 
241
        my_store.register_suffix('gz')
276
242
        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']))
 
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']))
281
246
 
282
247
    def test_add_simple(self):
283
248
        stream = StringIO("content")
297
262
        my_store.register_suffix('dsc')
298
263
        my_store.add(stream, "foo", 'dsc')
299
264
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
300
 
 
 
265
        
301
266
    def test_add_simple_suffixed(self):
302
267
        stream = StringIO("content")
303
268
        my_store = InstrumentedTransportStore(MockTransport(), True)
305
270
        my_store.add(stream, "foo", 'dsc')
306
271
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
307
272
 
308
 
    def get_populated_store(self, prefixed=False,
309
 
            store_class=TextStore, compressed=False):
310
 
        my_store = store_class(MemoryTransport(), prefixed,
311
 
                               compressed=compressed)
 
273
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
274
        my_store = store_class(MemoryTransport(), prefixed)
312
275
        my_store.register_suffix('sig')
313
276
        stream = StringIO("signature")
314
277
        my_store.add(stream, "foo", 'sig')
317
280
        stream = StringIO("signature for missing base")
318
281
        my_store.add(stream, "missing", 'sig')
319
282
        return my_store
320
 
 
 
283
        
321
284
    def test_has_simple(self):
322
285
        my_store = self.get_populated_store()
323
286
        self.assertEqual(True, my_store.has_id('foo'))
357
320
                         my_store.get('missing', 'sig').read())
358
321
 
359
322
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(),
361
 
                             prefixed=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(),
387
 
                             prefixed=True, compressed=True)
 
348
        to_store = CompressedTextStore(MemoryTransport(), True)
388
349
        to_store.register_suffix('sig')
389
 
        to_store.copy_all_ids(from_store)
 
350
        copy_all(from_store, to_store)
390
351
        self.assertEqual(1, len(to_store))
391
352
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
392
353
        self.assertEqual('content', to_store.get('foo').read())
393
354
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
394
355
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
395
 
 
396
 
    def test_relpath_escaped(self):
397
 
        my_store = TransportStore(MemoryTransport())
398
 
        self.assertEqual('%25', my_store._relpath('%'))
399
 
 
400
 
    def test_escaped_uppercase(self):
401
 
        """Uppercase letters are escaped for safety on Windows"""
402
 
        my_store = TransportStore(MemoryTransport(), prefixed=True,
403
 
            escaped=True)
404
 
        # a particularly perverse file-id! :-)
405
 
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
406
 
 
407
 
 
408
 
class TestVersionFileStore(TestCaseWithTransport):
409
 
 
410
 
    def get_scope(self):
411
 
        return self._transaction
412
 
 
413
 
    def setUp(self):
414
 
        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
419
 
 
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', [], [])
431
 
 
432
 
    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
 
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
439
 
 
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))