← 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: Vincent Ladeuil
  • Date: 2012-08-02 14:14:36 UTC
  • mto: This revision was merged to the branch mainline in revision 6551.
  • Revision ID: v.ladeuil+lp@free.fr-20120802141436-pa7qqfq0qj3t4oup
Fix obvious bug, the filter stack needs the path to not be None.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
1
 
# Copyright (C) 2005 by Canonical Development Ltd
2
 
 
 
1
# Copyright (C) 2005-2009, 2011 Canonical 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Test Store implementations."""
18
18
 
20
20
import os
21
21
import gzip
22
22
 
23
 
from bzrlib.errors import BzrError, UnlistableStore
24
 
from bzrlib.store import copy_all
25
 
from bzrlib.transport.local import LocalTransport
26
 
from bzrlib.transport import NoSuchFile
 
23
import bzrlib.errors as errors
 
24
from bzrlib.errors import BzrError
 
25
from bzrlib.store import TransportStore
27
26
from bzrlib.store.text import TextStore
28
 
from bzrlib.selftest import TestCase, TestCaseInTempDir
29
 
import bzrlib.store as store
 
27
from bzrlib.store.versioned import VersionedFileStore
 
28
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
 
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
32
33
 
33
34
 
34
35
class TestStores(object):
48
49
        """Test copying"""
49
50
        os.mkdir('a')
50
51
        store_a = self.get_store('a')
51
 
        store_a.add('foo', '1')
 
52
        store_a.add(StringIO('foo'), '1')
52
53
        os.mkdir('b')
53
54
        store_b = self.get_store('b')
54
 
        copy_all(store_a, store_b)
 
55
        store_b.copy_all_ids(store_a)
55
56
        self.assertEqual(store_a.get('1').read(), 'foo')
56
57
        self.assertEqual(store_b.get('1').read(), 'foo')
57
58
        # TODO: Switch the exception form UnlistableStore to
63
64
    def test_get(self):
64
65
        store = self.get_store()
65
66
        self.fill_store(store)
66
 
    
 
67
 
67
68
        self.check_content(store, 'a', 'hello')
68
69
        self.check_content(store, 'b', 'other')
69
70
        self.check_content(store, 'c', 'something')
70
 
    
 
71
 
71
72
        # Make sure that requesting a non-existing file fails
72
73
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
73
74
 
80
81
 
81
82
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
83
 
83
 
    def get_store(self, path='.'):
84
 
        t = LocalTransport(path)
 
84
    def get_store(self, path=u'.'):
 
85
        t = transport.get_transport_from_path(path)
85
86
        return TextStore(t, compressed=True)
86
87
 
87
88
    def test_total_size(self):
88
 
        store = self.get_store('.')
 
89
        store = self.get_store(u'.')
89
90
        store.register_suffix('dsc')
90
91
        store.add(StringIO('goodbye'), '123123')
91
92
        store.add(StringIO('goodbye2'), '123123', 'dsc')
92
93
        # these get gzipped - content should be stable
93
94
        self.assertEqual(store.total_size(), (2, 55))
94
 
        
 
95
 
95
96
    def test__relpath_suffixed(self):
96
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
97
        my_store = TextStore(MockTransport(),
 
98
                             prefixed=True, compressed=True)
97
99
        my_store.register_suffix('dsc')
98
100
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
99
101
 
100
102
 
101
103
class TestMemoryStore(TestCase):
102
 
    
 
104
 
103
105
    def get_store(self):
104
 
        return store.ImmutableMemoryStore()
105
 
    
106
 
    def test_imports(self):
107
 
        from bzrlib.store import ImmutableMemoryStore
 
106
        return TextStore(MemoryTransport())
108
107
 
109
108
    def test_add_and_retrieve(self):
110
109
        store = self.get_store()
117
116
 
118
117
    def test_missing_is_absent(self):
119
118
        store = self.get_store()
120
 
        self.failIf('aa' in store)
 
119
        self.assertFalse('aa' in store)
121
120
 
122
121
    def test_adding_fails_when_present(self):
123
122
        my_store = self.get_store()
139
138
 
140
139
class TestTextStore(TestCaseInTempDir, TestStores):
141
140
 
142
 
    def get_store(self, path='.'):
143
 
        t = LocalTransport(path)
 
141
    def get_store(self, path=u'.'):
 
142
        t = transport.get_transport_from_path(path)
144
143
        return TextStore(t, compressed=False)
145
144
 
146
145
    def test_total_size(self):
157
156
 
158
157
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
158
 
160
 
    def get_store(self, path='.', compressed=True):
161
 
        t = LocalTransport(path)
 
159
    def get_store(self, path=u'.', compressed=True):
 
160
        t = transport.get_transport_from_path(path)
162
161
        return TextStore(t, compressed=compressed)
163
162
 
164
163
    def test_get_mixed(self):
165
 
        cs = self.get_store('.', compressed=True)
166
 
        s = self.get_store('.', compressed=False)
 
164
        cs = self.get_store(u'.', compressed=True)
 
165
        s = self.get_store(u'.', compressed=False)
167
166
        cs.add(StringIO('hello there'), 'a')
168
167
 
169
 
        self.failUnlessExists('a.gz')
170
 
        self.failIf(os.path.lexists('a'))
 
168
        self.assertPathExists('a.gz')
 
169
        self.assertFalse(os.path.lexists('a'))
171
170
 
172
171
        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
173
172
 
175
174
        self.assertEquals(s.has_id('a'), True)
176
175
        self.assertEquals(cs.get('a').read(), 'hello there')
177
176
        self.assertEquals(s.get('a').read(), 'hello there')
178
 
        
 
177
 
179
178
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
180
179
 
181
180
        s.add(StringIO('goodbye'), 'b')
182
 
        self.failUnlessExists('b')
183
 
        self.failIf(os.path.lexists('b.gz'))
 
181
        self.assertPathExists('b')
 
182
        self.assertFalse(os.path.lexists('b.gz'))
184
183
        self.assertEquals(open('b').read(), 'goodbye')
185
184
 
186
185
        self.assertEquals(cs.has_id('b'), True)
187
186
        self.assertEquals(s.has_id('b'), True)
188
187
        self.assertEquals(cs.get('b').read(), 'goodbye')
189
188
        self.assertEquals(s.get('b').read(), 'goodbye')
190
 
        
 
189
 
191
190
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
192
191
 
193
192
class MockTransport(transport.Transport):
205
204
        return
206
205
 
207
206
 
208
 
class InstrumentedTransportStore(store.TransportStore):
 
207
class InstrumentedTransportStore(TransportStore):
209
208
    """An instrumented TransportStore.
210
209
 
211
210
    Here we replace template method worker methods with calls that record the
231
230
class TestMockTransport(TestCase):
232
231
 
233
232
    def test_isinstance(self):
234
 
        self.failUnless(isinstance(MockTransport(), transport.Transport))
 
233
        self.assertIsInstance(MockTransport(), transport.Transport)
235
234
 
236
235
    def test_has(self):
237
236
        self.assertEqual(False, MockTransport().has('foo'))
241
240
 
242
241
 
243
242
class TestTransportStore(TestCase):
244
 
    
 
243
 
245
244
    def test__relpath_invalid(self):
246
 
        my_store = store.TransportStore(MockTransport())
 
245
        my_store = TransportStore(MockTransport())
247
246
        self.assertRaises(ValueError, my_store._relpath, '/foo')
248
247
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
249
248
 
250
249
    def test_register_invalid_suffixes(self):
251
 
        my_store = store.TransportStore(MockTransport())
 
250
        my_store = TransportStore(MockTransport())
252
251
        self.assertRaises(ValueError, my_store.register_suffix, '/')
253
252
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
254
253
 
255
254
    def test__relpath_unregister_suffixes(self):
256
 
        my_store = store.TransportStore(MockTransport())
 
255
        my_store = TransportStore(MockTransport())
257
256
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
258
257
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
259
258
 
260
259
    def test__relpath_simple(self):
261
 
        my_store = store.TransportStore(MockTransport())
 
260
        my_store = TransportStore(MockTransport())
262
261
        self.assertEqual("foo", my_store._relpath('foo'))
263
262
 
264
263
    def test__relpath_prefixed(self):
265
 
        my_store = store.TransportStore(MockTransport(), True)
 
264
        my_store = TransportStore(MockTransport(), True)
266
265
        self.assertEqual('45/foo', my_store._relpath('foo'))
267
266
 
268
267
    def test__relpath_simple_suffixed(self):
269
 
        my_store = store.TransportStore(MockTransport())
 
268
        my_store = TransportStore(MockTransport())
270
269
        my_store.register_suffix('bar')
271
270
        my_store.register_suffix('baz')
272
271
        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
273
272
        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
274
273
 
275
274
    def test__relpath_prefixed_suffixed(self):
276
 
        my_store = store.TransportStore(MockTransport(), True)
 
275
        my_store = TransportStore(MockTransport(), True)
277
276
        my_store.register_suffix('bar')
278
277
        my_store.register_suffix('baz')
279
278
        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
298
297
        my_store.register_suffix('dsc')
299
298
        my_store.add(stream, "foo", 'dsc')
300
299
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
301
 
        
 
300
 
302
301
    def test_add_simple_suffixed(self):
303
302
        stream = StringIO("content")
304
303
        my_store = InstrumentedTransportStore(MockTransport(), True)
318
317
        stream = StringIO("signature for missing base")
319
318
        my_store.add(stream, "missing", 'sig')
320
319
        return my_store
321
 
        
 
320
 
322
321
    def test_has_simple(self):
323
322
        my_store = self.get_populated_store()
324
323
        self.assertEqual(True, my_store.has_id('foo'))
358
357
                         my_store.get('missing', 'sig').read())
359
358
 
360
359
    def test___iter__no_suffix(self):
361
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
360
        my_store = TextStore(MemoryTransport(),
 
361
                             prefixed=False, compressed=False)
362
362
        stream = StringIO("content")
363
363
        my_store.add(stream, "foo")
364
364
        self.assertEqual(set(['foo']),
383
383
 
384
384
    def test_copy_suffixes(self):
385
385
        from_store = self.get_populated_store()
386
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
386
        to_store = TextStore(MemoryTransport(),
 
387
                             prefixed=True, compressed=True)
387
388
        to_store.register_suffix('sig')
388
 
        copy_all(from_store, to_store)
 
389
        to_store.copy_all_ids(from_store)
389
390
        self.assertEqual(1, len(to_store))
390
391
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
391
392
        self.assertEqual('content', to_store.get('foo').read())
393
394
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
394
395
 
395
396
    def test_relpath_escaped(self):
396
 
        my_store = store.TransportStore(MemoryTransport())
 
397
        my_store = TransportStore(MemoryTransport())
397
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))