← 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: 2007-06-12 06:07:27 UTC
  • mfrom: (2522.1.1 bzr.dev)
  • Revision ID: pqm@pqm.ubuntu.com-20070612060727-v8nd5etbkay15fm2
prepare for 0.18 development

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 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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
20
20
import os
21
21
import gzip
22
22
 
23
 
from bzrlib.errors import BzrError, UnlistableStore
24
 
from bzrlib.store import copy_all
 
23
import bzrlib.errors as errors
 
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
25
25
from bzrlib.transport.local import LocalTransport
26
 
from bzrlib.transport import NoSuchFile
27
26
from bzrlib.store.text import TextStore
28
 
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
29
28
import bzrlib.store as store
 
29
import bzrlib.transactions as transactions
30
30
import bzrlib.transport as transport
31
31
from bzrlib.transport.memory import MemoryTransport
32
32
 
38
38
        f = store.get(fileid)
39
39
        self.assertEqual(f.read(), value)
40
40
 
 
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
 
41
49
    def fill_store(self, store):
42
50
        store.add(StringIO('hello'), 'a')
43
51
        store.add(StringIO('other'), 'b')
48
56
        """Test copying"""
49
57
        os.mkdir('a')
50
58
        store_a = self.get_store('a')
51
 
        store_a.add('foo', '1')
 
59
        store_a.add(StringIO('foo'), '1')
52
60
        os.mkdir('b')
53
61
        store_b = self.get_store('b')
54
 
        copy_all(store_a, store_b)
 
62
        store_b.copy_all_ids(store_a)
55
63
        self.assertEqual(store_a.get('1').read(), 'foo')
56
64
        self.assertEqual(store_b.get('1').read(), 'foo')
57
65
        # TODO: Switch the exception form UnlistableStore to
80
88
 
81
89
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
90
 
83
 
    def get_store(self, path='.'):
84
 
        t = LocalTransport(path)
 
91
    def get_store(self, path=u'.'):
 
92
        t = transport.get_transport(path)
85
93
        return TextStore(t, compressed=True)
86
94
 
87
95
    def test_total_size(self):
88
 
        store = self.get_store('.')
 
96
        store = self.get_store(u'.')
89
97
        store.register_suffix('dsc')
90
98
        store.add(StringIO('goodbye'), '123123')
91
99
        store.add(StringIO('goodbye2'), '123123', 'dsc')
93
101
        self.assertEqual(store.total_size(), (2, 55))
94
102
        
95
103
    def test__relpath_suffixed(self):
96
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
104
        my_store = TextStore(MockTransport(),
 
105
                             prefixed=True, compressed=True)
97
106
        my_store.register_suffix('dsc')
98
107
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
99
108
 
101
110
class TestMemoryStore(TestCase):
102
111
    
103
112
    def get_store(self):
104
 
        return store.ImmutableMemoryStore()
 
113
        return TextStore(MemoryTransport())
105
114
    
106
 
    def test_imports(self):
107
 
        from bzrlib.store import ImmutableMemoryStore
108
 
 
109
115
    def test_add_and_retrieve(self):
110
116
        store = self.get_store()
111
117
        store.add(StringIO('hello'), 'aa')
139
145
 
140
146
class TestTextStore(TestCaseInTempDir, TestStores):
141
147
 
142
 
    def get_store(self, path='.'):
143
 
        t = LocalTransport(path)
 
148
    def get_store(self, path=u'.'):
 
149
        t = transport.get_transport(path)
144
150
        return TextStore(t, compressed=False)
145
151
 
146
152
    def test_total_size(self):
157
163
 
158
164
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
165
 
160
 
    def get_store(self, path='.', compressed=True):
161
 
        t = LocalTransport(path)
 
166
    def get_store(self, path=u'.', compressed=True):
 
167
        t = transport.get_transport(path)
162
168
        return TextStore(t, compressed=compressed)
163
169
 
164
170
    def test_get_mixed(self):
165
 
        cs = self.get_store('.', compressed=True)
166
 
        s = self.get_store('.', compressed=False)
 
171
        cs = self.get_store(u'.', compressed=True)
 
172
        s = self.get_store(u'.', compressed=False)
167
173
        cs.add(StringIO('hello there'), 'a')
168
174
 
169
175
        self.failUnlessExists('a.gz')
358
364
                         my_store.get('missing', 'sig').read())
359
365
 
360
366
    def test___iter__no_suffix(self):
361
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
367
        my_store = TextStore(MemoryTransport(),
 
368
                             prefixed=False, compressed=False)
362
369
        stream = StringIO("content")
363
370
        my_store.add(stream, "foo")
364
371
        self.assertEqual(set(['foo']),
383
390
 
384
391
    def test_copy_suffixes(self):
385
392
        from_store = self.get_populated_store()
386
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
393
        to_store = TextStore(MemoryTransport(),
 
394
                             prefixed=True, compressed=True)
387
395
        to_store.register_suffix('sig')
388
 
        copy_all(from_store, to_store)
 
396
        to_store.copy_all_ids(from_store)
389
397
        self.assertEqual(1, len(to_store))
390
398
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
391
399
        self.assertEqual('content', to_store.get('foo').read())
395
403
    def test_relpath_escaped(self):
396
404
        my_store = store.TransportStore(MemoryTransport())
397
405
        self.assertEqual('%25', my_store._relpath('%'))
 
406
 
 
407
    def test_escaped_uppercase(self):
 
408
        """Uppercase letters are escaped for safety on Windows"""
 
409
        my_store = store.TransportStore(MemoryTransport(), escaped=True)
 
410
        # a particularly perverse file-id! :-)
 
411
        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
 
412
 
 
413
 
 
414
class TestVersionFileStore(TestCaseWithTransport):
 
415
 
 
416
    def setUp(self):
 
417
        super(TestVersionFileStore, self).setUp()
 
418
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
419
 
 
420
    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)
 
436
 
 
437
    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)
 
443
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
 
444