1
# Copyright (C) 2005 by Canonical Development Ltd
1
# Copyright (C) 2005, 2007 Canonical 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
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
49
50
store_a = self.get_store('a')
50
store_a.add('foo', '1')
51
store_a.add(StringIO('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
80
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
83
def get_store(self, path=u'.'):
83
t = LocalTransport(path)
84
t = transport.get_transport(path)
84
85
return TextStore(t, compressed=True)
86
87
def test_total_size(self):
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']))
100
102
class TestMemoryStore(TestCase):
102
104
def get_store(self):
103
return store.ImmutableMemoryStore()
105
return TextStore(MemoryTransport())
105
def test_imports(self):
106
from bzrlib.store import ImmutableMemoryStore
108
107
def test_add_and_retrieve(self):
109
108
store = self.get_store()
110
109
store.add(StringIO('hello'), 'aa')
139
138
class TestTextStore(TestCaseInTempDir, TestStores):
141
140
def get_store(self, path=u'.'):
142
t = LocalTransport(path)
141
t = transport.get_transport(path)
143
142
return TextStore(t, compressed=False)
145
144
def test_total_size(self):
157
156
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
158
def get_store(self, path=u'.', compressed=True):
160
t = LocalTransport(path)
159
t = transport.get_transport(path)
161
160
return TextStore(t, compressed=compressed)
163
162
def test_get_mixed(self):
357
356
my_store.get('missing', 'sig').read())
359
358
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(), False, compressed=False)
359
my_store = TextStore(MemoryTransport(),
360
prefixed=False, compressed=False)
361
361
stream = StringIO("content")
362
362
my_store.add(stream, "foo")
363
363
self.assertEqual(set(['foo']),
383
383
def test_copy_suffixes(self):
384
384
from_store = self.get_populated_store()
385
to_store = TextStore(MemoryTransport(), True, compressed=True)
385
to_store = TextStore(MemoryTransport(),
386
prefixed=True, compressed=True)
386
387
to_store.register_suffix('sig')
387
copy_all(from_store, to_store)
388
to_store.copy_all_ids(from_store)
388
389
self.assertEqual(1, len(to_store))
389
390
self.assertEqual(set(['foo']), set(to_store.__iter__()))
390
391
self.assertEqual('content', to_store.get('foo').read())
394
395
def test_relpath_escaped(self):
395
396
my_store = store.TransportStore(MemoryTransport())
396
397
self.assertEqual('%25', my_store._relpath('%'))
399
def test_escaped_uppercase(self):
400
"""Uppercase letters are escaped for safety on Windows"""
401
my_store = store.TransportStore(MemoryTransport(), escaped=True)
402
# a particularly perverse file-id! :-)
403
self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
406
class TestVersionFileStore(TestCaseWithTransport):
409
super(TestVersionFileStore, self).setUp()
410
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
412
def test_get_weave_registers_dirty_in_write(self):
413
transaction = transactions.WriteTransaction()
414
vf = self.vfstore.get_weave_or_empty('id', transaction)
416
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
417
transaction = transactions.WriteTransaction()
418
vf = self.vfstore.get_weave('id', transaction)
420
self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
422
def test_get_weave_or_empty_readonly_fails(self):
423
transaction = transactions.ReadOnlyTransaction()
424
vf = self.assertRaises(errors.ReadOnlyError,
425
self.vfstore.get_weave_or_empty,
429
def test_get_weave_readonly_cant_write(self):
430
transaction = transactions.WriteTransaction()
431
vf = self.vfstore.get_weave_or_empty('id', transaction)
433
transaction = transactions.ReadOnlyTransaction()
434
vf = self.vfstore.get_weave_or_empty('id', transaction)
435
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])