23
23
import bzrlib.errors as errors
24
from bzrlib.errors import BzrError
25
from bzrlib.store import TransportStore
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
25
from bzrlib.transport.local import LocalTransport
26
26
from bzrlib.store.text import TextStore
27
from bzrlib.store.versioned import VersionedFileStore
28
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
import bzrlib.store as store
29
import bzrlib.store.versioned
29
30
import bzrlib.transactions as transactions
30
31
import bzrlib.transport as transport
31
32
from bzrlib.transport.memory import MemoryTransport
32
from bzrlib.weave import WeaveFile
35
35
class TestStores(object):
165
165
s = self.get_store(u'.', compressed=False)
166
166
cs.add(StringIO('hello there'), 'a')
168
self.assertPathExists('a.gz')
169
self.assertFalse(os.path.lexists('a'))
168
self.failUnlessExists('a.gz')
169
self.failIf(os.path.lexists('a'))
171
171
self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
178
178
self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
180
180
s.add(StringIO('goodbye'), 'b')
181
self.assertPathExists('b')
182
self.assertFalse(os.path.lexists('b.gz'))
181
self.failUnlessExists('b')
182
self.failIf(os.path.lexists('b.gz'))
183
183
self.assertEquals(open('b').read(), 'goodbye')
185
185
self.assertEquals(cs.has_id('b'), True)
242
242
class TestTransportStore(TestCase):
244
244
def test__relpath_invalid(self):
245
my_store = TransportStore(MockTransport())
245
my_store = store.TransportStore(MockTransport())
246
246
self.assertRaises(ValueError, my_store._relpath, '/foo')
247
247
self.assertRaises(ValueError, my_store._relpath, 'foo/')
249
249
def test_register_invalid_suffixes(self):
250
my_store = TransportStore(MockTransport())
250
my_store = store.TransportStore(MockTransport())
251
251
self.assertRaises(ValueError, my_store.register_suffix, '/')
252
252
self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
254
254
def test__relpath_unregister_suffixes(self):
255
my_store = TransportStore(MockTransport())
255
my_store = store.TransportStore(MockTransport())
256
256
self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
257
257
self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
259
259
def test__relpath_simple(self):
260
my_store = TransportStore(MockTransport())
260
my_store = store.TransportStore(MockTransport())
261
261
self.assertEqual("foo", my_store._relpath('foo'))
263
263
def test__relpath_prefixed(self):
264
my_store = TransportStore(MockTransport(), True)
264
my_store = store.TransportStore(MockTransport(), True)
265
265
self.assertEqual('45/foo', my_store._relpath('foo'))
267
267
def test__relpath_simple_suffixed(self):
268
my_store = TransportStore(MockTransport())
268
my_store = store.TransportStore(MockTransport())
269
269
my_store.register_suffix('bar')
270
270
my_store.register_suffix('baz')
271
271
self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
272
272
self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
274
274
def test__relpath_prefixed_suffixed(self):
275
my_store = TransportStore(MockTransport(), True)
275
my_store = store.TransportStore(MockTransport(), True)
276
276
my_store.register_suffix('bar')
277
277
my_store.register_suffix('baz')
278
278
self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
394
394
self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
396
396
def test_relpath_escaped(self):
397
my_store = TransportStore(MemoryTransport())
397
my_store = store.TransportStore(MemoryTransport())
398
398
self.assertEqual('%25', my_store._relpath('%'))
400
400
def test_escaped_uppercase(self):
401
401
"""Uppercase letters are escaped for safety on Windows"""
402
my_store = TransportStore(MemoryTransport(), prefixed=True,
402
my_store = store.TransportStore(MemoryTransport(), prefixed=True,
404
404
# a particularly perverse file-id! :-)
405
405
self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
438
437
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
440
439
def test___iter__escaped(self):
441
self.vfstore = VersionedFileStore(MemoryTransport(),
442
prefixed=True, escaped=True, versionedfile_class=WeaveFile)
440
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
441
prefixed=True, escaped=True)
443
442
self.vfstore.get_scope = self.get_scope
444
443
self._transaction = transactions.WriteTransaction()
445
444
vf = self.vfstore.get_weave_or_empty(' ', self._transaction)