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
import bzrlib.store.versioned
30
29
import bzrlib.transactions as transactions
31
30
import bzrlib.transport as transport
32
31
from bzrlib.transport.memory import MemoryTransport
32
from bzrlib.weave import WeaveFile
35
35
class TestStores(object):
82
82
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
84
84
def get_store(self, path=u'.'):
85
t = transport.get_transport(path)
85
t = transport.get_transport_from_path(path)
86
86
return TextStore(t, compressed=True)
88
88
def test_total_size(self):
139
139
class TestTextStore(TestCaseInTempDir, TestStores):
141
141
def get_store(self, path=u'.'):
142
t = transport.get_transport(path)
142
t = transport.get_transport_from_path(path)
143
143
return TextStore(t, compressed=False)
145
145
def test_total_size(self):
157
157
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
159
def get_store(self, path=u'.', compressed=True):
160
t = transport.get_transport(path)
160
t = transport.get_transport_from_path(path)
161
161
return TextStore(t, compressed=compressed)
163
163
def test_get_mixed(self):
165
165
s = self.get_store(u'.', compressed=False)
166
166
cs.add(StringIO('hello there'), 'a')
168
self.failUnlessExists('a.gz')
169
self.failIf(os.path.lexists('a'))
168
self.assertPathExists('a.gz')
169
self.assertFalse(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.failUnlessExists('b')
182
self.failIf(os.path.lexists('b.gz'))
181
self.assertPathExists('b')
182
self.assertFalse(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 = store.TransportStore(MockTransport())
245
my_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 = store.TransportStore(MockTransport())
250
my_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 = store.TransportStore(MockTransport())
255
my_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 = store.TransportStore(MockTransport())
260
my_store = TransportStore(MockTransport())
261
261
self.assertEqual("foo", my_store._relpath('foo'))
263
263
def test__relpath_prefixed(self):
264
my_store = store.TransportStore(MockTransport(), True)
264
my_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 = store.TransportStore(MockTransport())
268
my_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 = store.TransportStore(MockTransport(), True)
275
my_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 = store.TransportStore(MemoryTransport())
397
my_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 = store.TransportStore(MemoryTransport(), prefixed=True,
402
my_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')
437
438
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
439
440
def test___iter__escaped(self):
440
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
441
prefixed=True, escaped=True)
441
self.vfstore = VersionedFileStore(MemoryTransport(),
442
prefixed=True, escaped=True, versionedfile_class=WeaveFile)
442
443
self.vfstore.get_scope = self.get_scope
443
444
self._transaction = transactions.WriteTransaction()
444
445
vf = self.vfstore.get_weave_or_empty(' ', self._transaction)