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
387
388
to_store = TextStore(MemoryTransport(),
388
389
prefixed=True, compressed=True)
389
390
to_store.register_suffix('sig')
390
copy_all(from_store, to_store)
391
to_store.copy_all_ids(from_store)
391
392
self.assertEqual(1, len(to_store))
392
393
self.assertEqual(set(['foo']), set(to_store.__iter__()))
393
394
self.assertEqual('content', to_store.get('foo').read())
397
398
def test_relpath_escaped(self):
398
399
my_store = store.TransportStore(MemoryTransport())
399
400
self.assertEqual('%25', my_store._relpath('%'))
403
class TestVersionFileStore(TestCaseWithTransport):
406
super(TestVersionFileStore, self).setUp()
407
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
409
def test_get_weave_registers_dirty_in_write(self):
410
transaction = transactions.WriteTransaction()
411
vf = self.vfstore.get_weave_or_empty('id', transaction)
413
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
414
transaction = transactions.WriteTransaction()
415
vf = self.vfstore.get_weave('id', transaction)
417
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
419
def test_get_weave_or_empty_readonly_fails(self):
420
transaction = transactions.ReadOnlyTransaction()
421
vf = self.assertRaises(errors.ReadOnlyError,
422
self.vfstore.get_weave_or_empty,
426
def test_get_weave_readonly_cant_write(self):
427
transaction = transactions.WriteTransaction()
428
vf = self.vfstore.get_weave_or_empty('id', transaction)
430
transaction = transactions.ReadOnlyTransaction()
431
vf = self.vfstore.get_weave_or_empty('id', transaction)
432
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])