← 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: John Arbash Meinel
  • Date: 2009-08-03 16:54:36 UTC
  • mto: This revision was merged to the branch mainline in revision 4591.
  • Revision ID: john@arbash-meinel.com-20090803165436-l7ps9z3y3oridzce
Use a cast to improve compiler conformance.

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
 
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
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
55
        store_b.copy_all_ids(store_a)
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
 
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
97
        my_store = TextStore(MockTransport(),
97
98
                             prefixed=True, compressed=True)
100
101
 
101
102
 
102
103
class TestMemoryStore(TestCase):
103
 
    
 
104
 
104
105
    def get_store(self):
105
106
        return TextStore(MemoryTransport())
106
 
    
 
107
 
107
108
    def test_add_and_retrieve(self):
108
109
        store = self.get_store()
109
110
        store.add(StringIO('hello'), 'aa')
173
174
        self.assertEquals(s.has_id('a'), True)
174
175
        self.assertEquals(cs.get('a').read(), 'hello there')
175
176
        self.assertEquals(s.get('a').read(), 'hello there')
176
 
        
 
177
 
177
178
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
178
179
 
179
180
        s.add(StringIO('goodbye'), 'b')
185
186
        self.assertEquals(s.has_id('b'), True)
186
187
        self.assertEquals(cs.get('b').read(), 'goodbye')
187
188
        self.assertEquals(s.get('b').read(), 'goodbye')
188
 
        
 
189
 
189
190
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
190
191
 
191
192
class MockTransport(transport.Transport):
239
240
 
240
241
 
241
242
class TestTransportStore(TestCase):
242
 
    
 
243
 
243
244
    def test__relpath_invalid(self):
244
245
        my_store = store.TransportStore(MockTransport())
245
246
        self.assertRaises(ValueError, my_store._relpath, '/foo')
296
297
        my_store.register_suffix('dsc')
297
298
        my_store.add(stream, "foo", 'dsc')
298
299
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
299
 
        
 
300
 
300
301
    def test_add_simple_suffixed(self):
301
302
        stream = StringIO("content")
302
303
        my_store = InstrumentedTransportStore(MockTransport(), True)
316
317
        stream = StringIO("signature for missing base")
317
318
        my_store.add(stream, "missing", 'sig')
318
319
        return my_store
319
 
        
 
320
 
320
321
    def test_has_simple(self):
321
322
        my_store = self.get_populated_store()
322
323
        self.assertEqual(True, my_store.has_id('foo'))
398
399
 
399
400
    def test_escaped_uppercase(self):
400
401
        """Uppercase letters are escaped for safety on Windows"""
401
 
        my_store = store.TransportStore(MemoryTransport(), escaped=True)
 
402
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
 
403
            escaped=True)
402
404
        # a particularly perverse file-id! :-)
403
 
        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
 
405
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
404
406
 
405
407
 
406
408
class TestVersionFileStore(TestCaseWithTransport):
407
409
 
 
410
    def get_scope(self):
 
411
        return self._transaction
 
412
 
408
413
    def setUp(self):
409
414
        super(TestVersionFileStore, self).setUp()
410
415
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
416
        self.vfstore.get_scope = self.get_scope
 
417
        self._transaction = None
411
418
 
412
419
    def test_get_weave_registers_dirty_in_write(self):
413
 
        transaction = transactions.WriteTransaction()
414
 
        vf = self.vfstore.get_weave_or_empty('id', transaction)
415
 
        transaction.finish()
416
 
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
417
 
        transaction = transactions.WriteTransaction()
418
 
        vf = self.vfstore.get_weave('id', transaction)
419
 
        transaction.finish()
420
 
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
421
 
 
422
 
    def test_get_weave_or_empty_readonly_fails(self):
423
 
        transaction = transactions.ReadOnlyTransaction()
424
 
        vf = self.assertRaises(errors.ReadOnlyError,
425
 
                               self.vfstore.get_weave_or_empty,
426
 
                               'id',
427
 
                               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', [], [])
428
430
 
429
431
    def test_get_weave_readonly_cant_write(self):
430
 
        transaction = transactions.WriteTransaction()
431
 
        vf = self.vfstore.get_weave_or_empty('id', transaction)
432
 
        transaction.finish()
433
 
        transaction = transactions.ReadOnlyTransaction()
434
 
        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)
435
437
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
436
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))