← 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: Canonical.com Patch Queue Manager
  • Date: 2006-05-19 06:14:38 UTC
  • mfrom: (1704.2.23 bzr.mbp.integration)
  • Revision ID: pqm@pqm.ubuntu.com-20060519061438-6300caf3926c3cff
(mbp) small fixes

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
1
 
# Copyright (C) 2005, 2007 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
 
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
30
29
import bzrlib.transactions as transactions
31
30
import bzrlib.transport as transport
32
31
from bzrlib.transport.memory import MemoryTransport
49
48
        """Test copying"""
50
49
        os.mkdir('a')
51
50
        store_a = self.get_store('a')
52
 
        store_a.add(StringIO('foo'), '1')
 
51
        store_a.add('foo', '1')
53
52
        os.mkdir('b')
54
53
        store_b = self.get_store('b')
55
54
        store_b.copy_all_ids(store_a)
64
63
    def test_get(self):
65
64
        store = self.get_store()
66
65
        self.fill_store(store)
67
 
 
 
66
    
68
67
        self.check_content(store, 'a', 'hello')
69
68
        self.check_content(store, 'b', 'other')
70
69
        self.check_content(store, 'c', 'something')
71
 
 
 
70
    
72
71
        # Make sure that requesting a non-existing file fails
73
72
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
74
73
 
82
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
83
82
 
84
83
    def get_store(self, path=u'.'):
85
 
        t = transport.get_transport(path)
 
84
        t = LocalTransport(path)
86
85
        return TextStore(t, compressed=True)
87
86
 
88
87
    def test_total_size(self):
92
91
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
92
        # these get gzipped - content should be stable
94
93
        self.assertEqual(store.total_size(), (2, 55))
95
 
 
 
94
        
96
95
    def test__relpath_suffixed(self):
97
96
        my_store = TextStore(MockTransport(),
98
97
                             prefixed=True, compressed=True)
101
100
 
102
101
 
103
102
class TestMemoryStore(TestCase):
104
 
 
 
103
    
105
104
    def get_store(self):
106
105
        return TextStore(MemoryTransport())
107
 
 
 
106
    
108
107
    def test_add_and_retrieve(self):
109
108
        store = self.get_store()
110
109
        store.add(StringIO('hello'), 'aa')
139
138
class TestTextStore(TestCaseInTempDir, TestStores):
140
139
 
141
140
    def get_store(self, path=u'.'):
142
 
        t = transport.get_transport(path)
 
141
        t = LocalTransport(path)
143
142
        return TextStore(t, compressed=False)
144
143
 
145
144
    def test_total_size(self):
157
156
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
157
 
159
158
    def get_store(self, path=u'.', compressed=True):
160
 
        t = transport.get_transport(path)
 
159
        t = LocalTransport(path)
161
160
        return TextStore(t, compressed=compressed)
162
161
 
163
162
    def test_get_mixed(self):
174
173
        self.assertEquals(s.has_id('a'), True)
175
174
        self.assertEquals(cs.get('a').read(), 'hello there')
176
175
        self.assertEquals(s.get('a').read(), 'hello there')
177
 
 
 
176
        
178
177
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
179
178
 
180
179
        s.add(StringIO('goodbye'), 'b')
186
185
        self.assertEquals(s.has_id('b'), True)
187
186
        self.assertEquals(cs.get('b').read(), 'goodbye')
188
187
        self.assertEquals(s.get('b').read(), 'goodbye')
189
 
 
 
188
        
190
189
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
191
190
 
192
191
class MockTransport(transport.Transport):
240
239
 
241
240
 
242
241
class TestTransportStore(TestCase):
243
 
 
 
242
    
244
243
    def test__relpath_invalid(self):
245
244
        my_store = store.TransportStore(MockTransport())
246
245
        self.assertRaises(ValueError, my_store._relpath, '/foo')
297
296
        my_store.register_suffix('dsc')
298
297
        my_store.add(stream, "foo", 'dsc')
299
298
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
300
 
 
 
299
        
301
300
    def test_add_simple_suffixed(self):
302
301
        stream = StringIO("content")
303
302
        my_store = InstrumentedTransportStore(MockTransport(), True)
317
316
        stream = StringIO("signature for missing base")
318
317
        my_store.add(stream, "missing", 'sig')
319
318
        return my_store
320
 
 
 
319
        
321
320
    def test_has_simple(self):
322
321
        my_store = self.get_populated_store()
323
322
        self.assertEqual(True, my_store.has_id('foo'))
399
398
 
400
399
    def test_escaped_uppercase(self):
401
400
        """Uppercase letters are escaped for safety on Windows"""
402
 
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
403
 
            escaped=True)
 
401
        my_store = store.TransportStore(MemoryTransport(), escaped=True)
404
402
        # a particularly perverse file-id! :-)
405
 
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
 
403
        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
406
404
 
407
405
 
408
406
class TestVersionFileStore(TestCaseWithTransport):
409
407
 
410
 
    def get_scope(self):
411
 
        return self._transaction
412
 
 
413
408
    def setUp(self):
414
409
        super(TestVersionFileStore, self).setUp()
415
410
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
416
 
        self.vfstore.get_scope = self.get_scope
417
 
        self._transaction = None
418
411
 
419
412
    def test_get_weave_registers_dirty_in_write(self):
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', [], [])
 
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)
430
428
 
431
429
    def test_get_weave_readonly_cant_write(self):
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)
 
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)
437
435
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
438
436
 
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))