← 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: 2008-10-31 04:39:04 UTC
  • mfrom: (3565.6.16 switch_nick)
  • Revision ID: pqm@pqm.ubuntu.com-20081031043904-52fnbfrloojemvcc
(mbp) branch nickname documentation

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, 2007 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
import bzrlib.errors as errors
23
24
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
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
28
import bzrlib.store as store
 
29
import bzrlib.store.versioned
 
30
import bzrlib.transactions as transactions
29
31
import bzrlib.transport as transport
30
32
from bzrlib.transport.memory import MemoryTransport
31
33
 
47
49
        """Test copying"""
48
50
        os.mkdir('a')
49
51
        store_a = self.get_store('a')
50
 
        store_a.add('foo', '1')
 
52
        store_a.add(StringIO('foo'), '1')
51
53
        os.mkdir('b')
52
54
        store_b = self.get_store('b')
53
 
        copy_all(store_a, store_b)
 
55
        store_b.copy_all_ids(store_a)
54
56
        self.assertEqual(store_a.get('1').read(), 'foo')
55
57
        self.assertEqual(store_b.get('1').read(), 'foo')
56
58
        # TODO: Switch the exception form UnlistableStore to
80
82
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
81
83
 
82
84
    def get_store(self, path=u'.'):
83
 
        t = LocalTransport(path)
 
85
        t = transport.get_transport(path)
84
86
        return TextStore(t, compressed=True)
85
87
 
86
88
    def test_total_size(self):
92
94
        self.assertEqual(store.total_size(), (2, 55))
93
95
        
94
96
    def test__relpath_suffixed(self):
95
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
97
        my_store = TextStore(MockTransport(),
 
98
                             prefixed=True, compressed=True)
96
99
        my_store.register_suffix('dsc')
97
100
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
98
101
 
100
103
class TestMemoryStore(TestCase):
101
104
    
102
105
    def get_store(self):
103
 
        return store.ImmutableMemoryStore()
 
106
        return TextStore(MemoryTransport())
104
107
    
105
 
    def test_imports(self):
106
 
        from bzrlib.store import ImmutableMemoryStore
107
 
 
108
108
    def test_add_and_retrieve(self):
109
109
        store = self.get_store()
110
110
        store.add(StringIO('hello'), 'aa')
139
139
class TestTextStore(TestCaseInTempDir, TestStores):
140
140
 
141
141
    def get_store(self, path=u'.'):
142
 
        t = LocalTransport(path)
 
142
        t = transport.get_transport(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 = LocalTransport(path)
 
160
        t = transport.get_transport(path)
161
161
        return TextStore(t, compressed=compressed)
162
162
 
163
163
    def test_get_mixed(self):
357
357
                         my_store.get('missing', 'sig').read())
358
358
 
359
359
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
360
        my_store = TextStore(MemoryTransport(),
 
361
                             prefixed=False, compressed=False)
361
362
        stream = StringIO("content")
362
363
        my_store.add(stream, "foo")
363
364
        self.assertEqual(set(['foo']),
382
383
 
383
384
    def test_copy_suffixes(self):
384
385
        from_store = self.get_populated_store()
385
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
386
        to_store = TextStore(MemoryTransport(),
 
387
                             prefixed=True, compressed=True)
386
388
        to_store.register_suffix('sig')
387
 
        copy_all(from_store, to_store)
 
389
        to_store.copy_all_ids(from_store)
388
390
        self.assertEqual(1, len(to_store))
389
391
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
390
392
        self.assertEqual('content', to_store.get('foo').read())
394
396
    def test_relpath_escaped(self):
395
397
        my_store = store.TransportStore(MemoryTransport())
396
398
        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))