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
92
93
self.assertEqual(store.total_size(), (2, 55))
94
95
def test__relpath_suffixed(self):
95
my_store = TextStore(MockTransport(), True, compressed=True)
96
my_store = TextStore(MockTransport(),
97
prefixed=True, compressed=True)
96
98
my_store.register_suffix('dsc')
97
99
self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
357
359
my_store.get('missing', 'sig').read())
359
361
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(), False, compressed=False)
362
my_store = TextStore(MemoryTransport(),
363
prefixed=False, compressed=False)
361
364
stream = StringIO("content")
362
365
my_store.add(stream, "foo")
363
366
self.assertEqual(set(['foo']),
383
386
def test_copy_suffixes(self):
384
387
from_store = self.get_populated_store()
385
to_store = TextStore(MemoryTransport(), True, compressed=True)
388
to_store = TextStore(MemoryTransport(),
389
prefixed=True, compressed=True)
386
390
to_store.register_suffix('sig')
387
copy_all(from_store, to_store)
391
to_store.copy_all_ids(from_store)
388
392
self.assertEqual(1, len(to_store))
389
393
self.assertEqual(set(['foo']), set(to_store.__iter__()))
390
394
self.assertEqual('content', to_store.get('foo').read())
394
398
def test_relpath_escaped(self):
395
399
my_store = store.TransportStore(MemoryTransport())
396
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', [], [])