23
23
import bzrlib.errors as errors
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
25
from bzrlib.transport.local import LocalTransport
24
from bzrlib.errors import BzrError
25
from bzrlib.store import TransportStore
26
26
from bzrlib.store.text import TextStore
27
from bzrlib.store.versioned import VersionedFileStore
27
28
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
import bzrlib.store as store
29
29
import bzrlib.transactions as transactions
30
30
import bzrlib.transport as transport
31
31
from bzrlib.transport.memory import MemoryTransport
32
from bzrlib.weave import WeaveFile
34
35
class TestStores(object):
181
174
self.assertEquals(s.has_id('a'), True)
182
175
self.assertEquals(cs.get('a').read(), 'hello there')
183
176
self.assertEquals(s.get('a').read(), 'hello there')
185
178
self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
187
180
s.add(StringIO('goodbye'), 'b')
188
self.failUnlessExists('b')
189
self.failIf(os.path.lexists('b.gz'))
181
self.assertPathExists('b')
182
self.assertFalse(os.path.lexists('b.gz'))
190
183
self.assertEquals(open('b').read(), 'goodbye')
192
185
self.assertEquals(cs.has_id('b'), True)
193
186
self.assertEquals(s.has_id('b'), True)
194
187
self.assertEquals(cs.get('b').read(), 'goodbye')
195
188
self.assertEquals(s.get('b').read(), 'goodbye')
197
190
self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
199
192
class MockTransport(transport.Transport):
249
242
class TestTransportStore(TestCase):
251
244
def test__relpath_invalid(self):
252
my_store = store.TransportStore(MockTransport())
245
my_store = TransportStore(MockTransport())
253
246
self.assertRaises(ValueError, my_store._relpath, '/foo')
254
247
self.assertRaises(ValueError, my_store._relpath, 'foo/')
256
249
def test_register_invalid_suffixes(self):
257
my_store = store.TransportStore(MockTransport())
250
my_store = TransportStore(MockTransport())
258
251
self.assertRaises(ValueError, my_store.register_suffix, '/')
259
252
self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
261
254
def test__relpath_unregister_suffixes(self):
262
my_store = store.TransportStore(MockTransport())
255
my_store = TransportStore(MockTransport())
263
256
self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
264
257
self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
266
259
def test__relpath_simple(self):
267
my_store = store.TransportStore(MockTransport())
260
my_store = TransportStore(MockTransport())
268
261
self.assertEqual("foo", my_store._relpath('foo'))
270
263
def test__relpath_prefixed(self):
271
my_store = store.TransportStore(MockTransport(), True)
264
my_store = TransportStore(MockTransport(), True)
272
265
self.assertEqual('45/foo', my_store._relpath('foo'))
274
267
def test__relpath_simple_suffixed(self):
275
my_store = store.TransportStore(MockTransport())
268
my_store = TransportStore(MockTransport())
276
269
my_store.register_suffix('bar')
277
270
my_store.register_suffix('baz')
278
271
self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
279
272
self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
281
274
def test__relpath_prefixed_suffixed(self):
282
my_store = store.TransportStore(MockTransport(), True)
275
my_store = TransportStore(MockTransport(), True)
283
276
my_store.register_suffix('bar')
284
277
my_store.register_suffix('baz')
285
278
self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
401
394
self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
403
396
def test_relpath_escaped(self):
404
my_store = store.TransportStore(MemoryTransport())
397
my_store = TransportStore(MemoryTransport())
405
398
self.assertEqual('%25', my_store._relpath('%'))
407
400
def test_escaped_uppercase(self):
408
401
"""Uppercase letters are escaped for safety on Windows"""
409
my_store = store.TransportStore(MemoryTransport(), escaped=True)
402
my_store = TransportStore(MemoryTransport(), prefixed=True,
410
404
# a particularly perverse file-id! :-)
411
self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
405
self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
414
408
class TestVersionFileStore(TestCaseWithTransport):
411
return self._transaction
417
414
super(TestVersionFileStore, self).setUp()
418
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
415
self.vfstore = VersionedFileStore(MemoryTransport(),
416
versionedfile_class=WeaveFile)
417
self.vfstore.get_scope = self.get_scope
418
self._transaction = None
420
420
def test_get_weave_registers_dirty_in_write(self):
421
transaction = transactions.WriteTransaction()
422
vf = self.vfstore.get_weave_or_empty('id', transaction)
424
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
425
transaction = transactions.WriteTransaction()
426
vf = self.vfstore.get_weave('id', transaction)
428
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
430
def test_get_weave_or_empty_readonly_fails(self):
431
transaction = transactions.ReadOnlyTransaction()
432
vf = self.assertRaises(errors.ReadOnlyError,
433
self.vfstore.get_weave_or_empty,
421
self._transaction = transactions.WriteTransaction()
422
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
423
self._transaction.finish()
424
self._transaction = None
425
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
426
self._transaction = transactions.WriteTransaction()
427
vf = self.vfstore.get_weave('id', self._transaction)
428
self._transaction.finish()
429
self._transaction = None
430
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
437
432
def test_get_weave_readonly_cant_write(self):
438
transaction = transactions.WriteTransaction()
439
vf = self.vfstore.get_weave_or_empty('id', transaction)
441
transaction = transactions.ReadOnlyTransaction()
442
vf = self.vfstore.get_weave_or_empty('id', transaction)
433
self._transaction = transactions.WriteTransaction()
434
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
435
self._transaction.finish()
436
self._transaction = transactions.ReadOnlyTransaction()
437
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
443
438
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
440
def test___iter__escaped(self):
441
self.vfstore = VersionedFileStore(MemoryTransport(),
442
prefixed=True, escaped=True, versionedfile_class=WeaveFile)
443
self.vfstore.get_scope = self.get_scope
444
self._transaction = transactions.WriteTransaction()
445
vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
446
vf.add_lines('a', [], [])
448
self._transaction.finish()
449
self.assertEqual([' '], list(self.vfstore))