/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Aaron Bentley
  • Date: 2005-12-25 00:38:48 UTC
  • mto: (1185.67.11 bzr.revision-storage)
  • mto: This revision was merged to the branch mainline in revision 1550.
  • Revision ID: aaron.bentley@utoronto.ca-20051225003848-111ac71170cb2605
Renamed Branch.storage to Branch.repository

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
 
20
20
import os
21
21
import gzip
22
22
 
23
 
import bzrlib.errors as errors
24
23
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
 
24
from bzrlib.store import copy_all
25
25
from bzrlib.transport.local import LocalTransport
26
26
from bzrlib.store.text import TextStore
27
 
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir
28
28
import bzrlib.store as store
29
 
import bzrlib.store.versioned
30
 
import bzrlib.transactions as transactions
31
29
import bzrlib.transport as transport
32
30
from bzrlib.transport.memory import MemoryTransport
33
31
 
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
 
82
80
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
83
81
 
84
82
    def get_store(self, path=u'.'):
85
 
        t = transport.get_transport(path)
 
83
        t = LocalTransport(path)
86
84
        return TextStore(t, compressed=True)
87
85
 
88
86
    def test_total_size(self):
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 = TextStore(MockTransport(), True, compressed=True)
99
96
        my_store.register_suffix('dsc')
100
97
        self.assertEqual('45/foo.dsc', 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()
139
139
class TestTextStore(TestCaseInTempDir, TestStores):
140
140
 
141
141
    def get_store(self, path=u'.'):
142
 
        t = transport.get_transport(path)
 
142
        t = LocalTransport(path)
143
143
        return TextStore(t, compressed=False)
144
144
 
145
145
    def test_total_size(self):
157
157
class TestMixedTextStore(TestCaseInTempDir, TestStores):
158
158
 
159
159
    def get_store(self, path=u'.', compressed=True):
160
 
        t = transport.get_transport(path)
 
160
        t = LocalTransport(path)
161
161
        return TextStore(t, compressed=compressed)
162
162
 
163
163
    def test_get_mixed(self):
174
174
        self.assertEquals(s.has_id('a'), True)
175
175
        self.assertEquals(cs.get('a').read(), 'hello there')
176
176
        self.assertEquals(s.get('a').read(), 'hello there')
177
 
 
 
177
        
178
178
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
179
179
 
180
180
        s.add(StringIO('goodbye'), 'b')
186
186
        self.assertEquals(s.has_id('b'), True)
187
187
        self.assertEquals(cs.get('b').read(), 'goodbye')
188
188
        self.assertEquals(s.get('b').read(), 'goodbye')
189
 
 
 
189
        
190
190
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
191
191
 
192
192
class MockTransport(transport.Transport):
240
240
 
241
241
 
242
242
class TestTransportStore(TestCase):
243
 
 
 
243
    
244
244
    def test__relpath_invalid(self):
245
245
        my_store = store.TransportStore(MockTransport())
246
246
        self.assertRaises(ValueError, my_store._relpath, '/foo')
297
297
        my_store.register_suffix('dsc')
298
298
        my_store.add(stream, "foo", 'dsc')
299
299
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
300
 
 
 
300
        
301
301
    def test_add_simple_suffixed(self):
302
302
        stream = StringIO("content")
303
303
        my_store = InstrumentedTransportStore(MockTransport(), True)
317
317
        stream = StringIO("signature for missing base")
318
318
        my_store.add(stream, "missing", 'sig')
319
319
        return my_store
320
 
 
 
320
        
321
321
    def test_has_simple(self):
322
322
        my_store = self.get_populated_store()
323
323
        self.assertEqual(True, my_store.has_id('foo'))
357
357
                         my_store.get('missing', 'sig').read())
358
358
 
359
359
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(),
361
 
                             prefixed=False, compressed=False)
 
360
        my_store = TextStore(MemoryTransport(), False, compressed=False)
362
361
        stream = StringIO("content")
363
362
        my_store.add(stream, "foo")
364
363
        self.assertEqual(set(['foo']),
383
382
 
384
383
    def test_copy_suffixes(self):
385
384
        from_store = self.get_populated_store()
386
 
        to_store = TextStore(MemoryTransport(),
387
 
                             prefixed=True, compressed=True)
 
385
        to_store = TextStore(MemoryTransport(), True, compressed=True)
388
386
        to_store.register_suffix('sig')
389
 
        to_store.copy_all_ids(from_store)
 
387
        copy_all(from_store, to_store)
390
388
        self.assertEqual(1, len(to_store))
391
389
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
392
390
        self.assertEqual('content', to_store.get('foo').read())
396
394
    def test_relpath_escaped(self):
397
395
        my_store = store.TransportStore(MemoryTransport())
398
396
        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 = 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 = store.versioned.VersionedFileStore(MemoryTransport())
416
 
        self.vfstore.get_scope = self.get_scope
417
 
        self._transaction = None
418
 
 
419
 
    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', [], [])
430
 
 
431
 
    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)
437
 
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
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))