~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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.transactions as transactions
29
30
import bzrlib.transport as transport
30
31
from bzrlib.transport.memory import MemoryTransport
31
32
 
47
48
        """Test copying"""
48
49
        os.mkdir('a')
49
50
        store_a = self.get_store('a')
50
 
        store_a.add('foo', '1')
 
51
        store_a.add(StringIO('foo'), '1')
51
52
        os.mkdir('b')
52
53
        store_b = self.get_store('b')
53
 
        copy_all(store_a, store_b)
 
54
        store_b.copy_all_ids(store_a)
54
55
        self.assertEqual(store_a.get('1').read(), 'foo')
55
56
        self.assertEqual(store_b.get('1').read(), 'foo')
56
57
        # TODO: Switch the exception form UnlistableStore to
80
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
81
82
 
82
83
    def get_store(self, path=u'.'):
83
 
        t = LocalTransport(path)
 
84
        t = transport.get_transport(path)
84
85
        return TextStore(t, compressed=True)
85
86
 
86
87
    def test_total_size(self):
101
102
class TestMemoryStore(TestCase):
102
103
    
103
104
    def get_store(self):
104
 
        return store.ImmutableMemoryStore()
 
105
        return TextStore(MemoryTransport())
105
106
    
106
 
    def test_imports(self):
107
 
        from bzrlib.store import ImmutableMemoryStore
108
 
 
109
107
    def test_add_and_retrieve(self):
110
108
        store = self.get_store()
111
109
        store.add(StringIO('hello'), 'aa')
140
138
class TestTextStore(TestCaseInTempDir, TestStores):
141
139
 
142
140
    def get_store(self, path=u'.'):
143
 
        t = LocalTransport(path)
 
141
        t = transport.get_transport(path)
144
142
        return TextStore(t, compressed=False)
145
143
 
146
144
    def test_total_size(self):
158
156
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
157
 
160
158
    def get_store(self, path=u'.', compressed=True):
161
 
        t = LocalTransport(path)
 
159
        t = transport.get_transport(path)
162
160
        return TextStore(t, compressed=compressed)
163
161
 
164
162
    def test_get_mixed(self):
387
385
        to_store = TextStore(MemoryTransport(),
388
386
                             prefixed=True, compressed=True)
389
387
        to_store.register_suffix('sig')
390
 
        copy_all(from_store, to_store)
 
388
        to_store.copy_all_ids(from_store)
391
389
        self.assertEqual(1, len(to_store))
392
390
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
393
391
        self.assertEqual('content', to_store.get('foo').read())
397
395
    def test_relpath_escaped(self):
398
396
        my_store = store.TransportStore(MemoryTransport())
399
397
        self.assertEqual('%25', my_store._relpath('%'))
 
398
 
 
399
    def test_escaped_uppercase(self):
 
400
        """Uppercase letters are escaped for safety on Windows"""
 
401
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
 
402
            escaped=True)
 
403
        # a particularly perverse file-id! :-)
 
404
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
 
405
 
 
406
 
 
407
class TestVersionFileStore(TestCaseWithTransport):
 
408
 
 
409
    def get_scope(self):
 
410
        return self._transaction
 
411
 
 
412
    def setUp(self):
 
413
        super(TestVersionFileStore, self).setUp()
 
414
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
415
        self.vfstore.get_scope = self.get_scope
 
416
        self._transaction = None
 
417
 
 
418
    def test_get_weave_registers_dirty_in_write(self):
 
419
        self._transaction = transactions.WriteTransaction()
 
420
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
421
        self._transaction.finish()
 
422
        self._transaction = None
 
423
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
424
        self._transaction = transactions.WriteTransaction()
 
425
        vf = self.vfstore.get_weave('id', self._transaction)
 
426
        self._transaction.finish()
 
427
        self._transaction = None
 
428
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
429
 
 
430
    def test_get_weave_readonly_cant_write(self):
 
431
        self._transaction = transactions.WriteTransaction()
 
432
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
433
        self._transaction.finish()
 
434
        self._transaction = transactions.ReadOnlyTransaction()
 
435
        vf = self.vfstore.get_weave_or_empty('id', self._transaction)
 
436
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
 
437
 
 
438
    def test___iter__escaped(self):
 
439
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
 
440
            prefixed=True, escaped=True)
 
441
        self.vfstore.get_scope = self.get_scope
 
442
        self._transaction = transactions.WriteTransaction()
 
443
        vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
 
444
        vf.add_lines('a', [], [])
 
445
        del vf
 
446
        self._transaction.finish()
 
447
        self.assertEqual([' '], list(self.vfstore))