← 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: Robert Collins
  • Date: 2009-07-07 04:32:13 UTC
  • mto: This revision was merged to the branch mainline in revision 4524.
  • Revision ID: robertc@robertcollins.net-20090707043213-4hjjhgr40iq7gk2d
More informative assertions in xml serialisation.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
1
 
# Copyright (C) 2005 Canonical Ltd
 
1
# Copyright (C) 2005, 2007 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., 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
 
26
26
from bzrlib.store.text import TextStore
27
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
28
import bzrlib.store as store
 
29
import bzrlib.store.versioned
29
30
import bzrlib.transactions as transactions
30
31
import bzrlib.transport as transport
31
32
from bzrlib.transport.memory import MemoryTransport
38
39
        f = store.get(fileid)
39
40
        self.assertEqual(f.read(), value)
40
41
 
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
 
 
49
42
    def fill_store(self, store):
50
43
        store.add(StringIO('hello'), 'a')
51
44
        store.add(StringIO('other'), 'b')
71
64
    def test_get(self):
72
65
        store = self.get_store()
73
66
        self.fill_store(store)
74
 
    
 
67
 
75
68
        self.check_content(store, 'a', 'hello')
76
69
        self.check_content(store, 'b', 'other')
77
70
        self.check_content(store, 'c', 'something')
78
 
    
 
71
 
79
72
        # Make sure that requesting a non-existing file fails
80
73
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
81
74
 
99
92
        store.add(StringIO('goodbye2'), '123123', 'dsc')
100
93
        # these get gzipped - content should be stable
101
94
        self.assertEqual(store.total_size(), (2, 55))
102
 
        
 
95
 
103
96
    def test__relpath_suffixed(self):
104
97
        my_store = TextStore(MockTransport(),
105
98
                             prefixed=True, compressed=True)
108
101
 
109
102
 
110
103
class TestMemoryStore(TestCase):
111
 
    
 
104
 
112
105
    def get_store(self):
113
106
        return TextStore(MemoryTransport())
114
 
    
 
107
 
115
108
    def test_add_and_retrieve(self):
116
109
        store = self.get_store()
117
110
        store.add(StringIO('hello'), 'aa')
181
174
        self.assertEquals(s.has_id('a'), True)
182
175
        self.assertEquals(cs.get('a').read(), 'hello there')
183
176
        self.assertEquals(s.get('a').read(), 'hello there')
184
 
        
 
177
 
185
178
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
186
179
 
187
180
        s.add(StringIO('goodbye'), 'b')
193
186
        self.assertEquals(s.has_id('b'), True)
194
187
        self.assertEquals(cs.get('b').read(), 'goodbye')
195
188
        self.assertEquals(s.get('b').read(), 'goodbye')
196
 
        
 
189
 
197
190
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
198
191
 
199
192
class MockTransport(transport.Transport):
247
240
 
248
241
 
249
242
class TestTransportStore(TestCase):
250
 
    
 
243
 
251
244
    def test__relpath_invalid(self):
252
245
        my_store = store.TransportStore(MockTransport())
253
246
        self.assertRaises(ValueError, my_store._relpath, '/foo')
304
297
        my_store.register_suffix('dsc')
305
298
        my_store.add(stream, "foo", 'dsc')
306
299
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
307
 
        
 
300
 
308
301
    def test_add_simple_suffixed(self):
309
302
        stream = StringIO("content")
310
303
        my_store = InstrumentedTransportStore(MockTransport(), True)
324
317
        stream = StringIO("signature for missing base")
325
318
        my_store.add(stream, "missing", 'sig')
326
319
        return my_store
327
 
        
 
320
 
328
321
    def test_has_simple(self):
329
322
        my_store = self.get_populated_store()
330
323
        self.assertEqual(True, my_store.has_id('foo'))
406
399
 
407
400
    def test_escaped_uppercase(self):
408
401
        """Uppercase letters are escaped for safety on Windows"""
409
 
        my_store = store.TransportStore(MemoryTransport(), escaped=True)
 
402
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
 
403
            escaped=True)
410
404
        # a particularly perverse file-id! :-)
411
 
        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
 
405
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
412
406
 
413
407
 
414
408
class TestVersionFileStore(TestCaseWithTransport):
415
409
 
 
410
    def get_scope(self):
 
411
        return self._transaction
 
412
 
416
413
    def setUp(self):
417
414
        super(TestVersionFileStore, self).setUp()
418
415
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
416
        self.vfstore.get_scope = self.get_scope
 
417
        self._transaction = None
419
418
 
420
419
    def test_get_weave_registers_dirty_in_write(self):
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)
 
420
        self._transaction = transactions.WriteTransaction()
 
421
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
422
        self._transaction.finish()
 
423
        self._transaction = None
 
424
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
425
        self._transaction = transactions.WriteTransaction()
 
426
        vf = self.vfstore.get_weave('id', self._transaction)
 
427
        self._transaction.finish()
 
428
        self._transaction = None
 
429
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
436
430
 
437
431
    def test_get_weave_readonly_cant_write(self):
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)
 
432
        self._transaction = transactions.WriteTransaction()
 
433
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
434
        self._transaction.finish()
 
435
        self._transaction = transactions.ReadOnlyTransaction()
 
436
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
443
437
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
444
438
 
 
439
    def test___iter__escaped(self):
 
440
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
 
441
            prefixed=True, escaped=True)
 
442
        self.vfstore.get_scope = self.get_scope
 
443
        self._transaction = transactions.WriteTransaction()
 
444
        vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
 
445
        vf.add_lines('a', [], [])
 
446
        del vf
 
447
        self._transaction.finish()
 
448
        self.assertEqual([' '], list(self.vfstore))