1
# Copyright (C) 2005, 2007 Canonical Ltd
1
# Copyright (C) 2005 by Canonical Development Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
11
# GNU General Public License for more details.
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23
import bzrlib.errors as errors
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
23
from bzrlib.errors import BzrError, UnlistableStore
24
from bzrlib.store import copy_all
25
25
from bzrlib.transport.local import LocalTransport
26
from bzrlib.transport import NoSuchFile
26
27
from bzrlib.store.text import TextStore
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
from bzrlib.tests import TestCase, TestCaseInTempDir
28
29
import bzrlib.store as store
29
import bzrlib.store.versioned
30
import bzrlib.transactions as transactions
31
30
import bzrlib.transport as transport
32
31
from bzrlib.transport.memory import MemoryTransport
51
50
store_a = self.get_store('a')
52
store_a.add(StringIO('foo'), '1')
51
store_a.add('foo', '1')
54
53
store_b = self.get_store('b')
55
store_b.copy_all_ids(store_a)
54
copy_all(store_a, store_b)
56
55
self.assertEqual(store_a.get('1').read(), 'foo')
57
56
self.assertEqual(store_b.get('1').read(), 'foo')
58
57
# TODO: Switch the exception form UnlistableStore to
94
93
self.assertEqual(store.total_size(), (2, 55))
96
95
def test__relpath_suffixed(self):
97
my_store = TextStore(MockTransport(),
98
prefixed=True, compressed=True)
96
my_store = TextStore(MockTransport(), True, compressed=True)
99
97
my_store.register_suffix('dsc')
100
98
self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
157
158
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
160
def get_store(self, path=u'.', compressed=True):
160
t = transport.get_transport(path)
161
t = LocalTransport(path)
161
162
return TextStore(t, compressed=compressed)
163
164
def test_get_mixed(self):
357
358
my_store.get('missing', 'sig').read())
359
360
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(),
361
prefixed=False, compressed=False)
361
my_store = TextStore(MemoryTransport(), False, compressed=False)
362
362
stream = StringIO("content")
363
363
my_store.add(stream, "foo")
364
364
self.assertEqual(set(['foo']),
384
384
def test_copy_suffixes(self):
385
385
from_store = self.get_populated_store()
386
to_store = TextStore(MemoryTransport(),
387
prefixed=True, compressed=True)
386
to_store = TextStore(MemoryTransport(), True, compressed=True)
388
387
to_store.register_suffix('sig')
389
to_store.copy_all_ids(from_store)
388
copy_all(from_store, to_store)
390
389
self.assertEqual(1, len(to_store))
391
390
self.assertEqual(set(['foo']), set(to_store.__iter__()))
392
391
self.assertEqual('content', to_store.get('foo').read())
396
395
def test_relpath_escaped(self):
397
396
my_store = store.TransportStore(MemoryTransport())
398
397
self.assertEqual('%25', my_store._relpath('%'))
400
def test_escaped_uppercase(self):
401
"""Uppercase letters are escaped for safety on Windows"""
402
my_store = store.TransportStore(MemoryTransport(), prefixed=True,
404
# a particularly perverse file-id! :-)
405
self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
408
class TestVersionFileStore(TestCaseWithTransport):
411
return self._transaction
414
super(TestVersionFileStore, self).setUp()
415
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
416
self.vfstore.get_scope = self.get_scope
417
self._transaction = None
419
def test_get_weave_registers_dirty_in_write(self):
420
self._transaction = transactions.WriteTransaction()
421
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
422
self._transaction.finish()
423
self._transaction = None
424
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
425
self._transaction = transactions.WriteTransaction()
426
vf = self.vfstore.get_weave('id', self._transaction)
427
self._transaction.finish()
428
self._transaction = None
429
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
431
def test_get_weave_readonly_cant_write(self):
432
self._transaction = transactions.WriteTransaction()
433
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
434
self._transaction.finish()
435
self._transaction = transactions.ReadOnlyTransaction()
436
vf = self.vfstore.get_weave_or_empty('id', self._transaction)
437
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
439
def test___iter__escaped(self):
440
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport(),
441
prefixed=True, escaped=True)
442
self.vfstore.get_scope = self.get_scope
443
self._transaction = transactions.WriteTransaction()
444
vf = self.vfstore.get_weave_or_empty(' ', self._transaction)
445
vf.add_lines('a', [], [])
447
self._transaction.finish()
448
self.assertEqual([' '], list(self.vfstore))