1
1
# Copyright (C) 2005 by Canonical Development Ltd
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.
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.
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
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
50
51
store_a.add('foo', '1')
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):
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)
86
87
def test_total_size(self):
101
102
class TestMemoryStore(TestCase):
103
104
def get_store(self):
104
return store.ImmutableMemoryStore()
105
return TextStore(MemoryTransport())
106
def test_imports(self):
107
from bzrlib.store import ImmutableMemoryStore
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):
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)
146
144
def test_total_size(self):
158
156
class TestMixedTextStore(TestCaseInTempDir, TestStores):
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)
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('%'))
399
def test_escaped_uppercase(self):
400
"""Uppercase letters are escaped for safety on Windows"""
401
my_store = store.TransportStore(MemoryTransport(), escaped=True)
402
# a particularly perverse file-id! :-)
403
self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
406
class TestVersionFileStore(TestCaseWithTransport):
409
super(TestVersionFileStore, self).setUp()
410
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
412
def test_get_weave_registers_dirty_in_write(self):
413
transaction = transactions.WriteTransaction()
414
vf = self.vfstore.get_weave_or_empty('id', transaction)
416
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
417
transaction = transactions.WriteTransaction()
418
vf = self.vfstore.get_weave('id', transaction)
420
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
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,
429
def test_get_weave_readonly_cant_write(self):
430
transaction = transactions.WriteTransaction()
431
vf = self.vfstore.get_weave_or_empty('id', transaction)
433
transaction = transactions.ReadOnlyTransaction()
434
vf = self.vfstore.get_weave_or_empty('id', transaction)
435
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])