1
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
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
37
38
f = store.get(fileid)
38
39
self.assertEqual(f.read(), value)
41
def test_add_str_deprecated(self):
43
store = self.get_store('a')
44
self.callDeprecated(['Passing a string to Store.add'
45
' was deprecated in version 0.11.'],
46
store.add, 'foo', '1')
47
self.assertEqual('foo', store.get('1').read())
40
49
def fill_store(self, store):
41
50
store.add(StringIO('hello'), 'a')
42
51
store.add(StringIO('other'), 'b')
49
58
store_a = self.get_store('a')
50
store_a.add('foo', '1')
59
store_a.add(StringIO('foo'), '1')
52
61
store_b = self.get_store('b')
53
copy_all(store_a, store_b)
62
store_b.copy_all_ids(store_a)
54
63
self.assertEqual(store_a.get('1').read(), 'foo')
55
64
self.assertEqual(store_b.get('1').read(), 'foo')
56
65
# TODO: Switch the exception form UnlistableStore to
80
89
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
82
91
def get_store(self, path=u'.'):
83
t = LocalTransport(path)
92
t = transport.get_transport(path)
84
93
return TextStore(t, compressed=True)
86
95
def test_total_size(self):
92
101
self.assertEqual(store.total_size(), (2, 55))
94
103
def test__relpath_suffixed(self):
95
my_store = TextStore(MockTransport(), True, compressed=True)
104
my_store = TextStore(MockTransport(),
105
prefixed=True, compressed=True)
96
106
my_store.register_suffix('dsc')
97
107
self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
100
110
class TestMemoryStore(TestCase):
102
112
def get_store(self):
103
return store.ImmutableMemoryStore()
113
return TextStore(MemoryTransport())
105
def test_imports(self):
106
from bzrlib.store import ImmutableMemoryStore
108
115
def test_add_and_retrieve(self):
109
116
store = self.get_store()
110
117
store.add(StringIO('hello'), 'aa')
139
146
class TestTextStore(TestCaseInTempDir, TestStores):
141
148
def get_store(self, path=u'.'):
142
t = LocalTransport(path)
149
t = transport.get_transport(path)
143
150
return TextStore(t, compressed=False)
145
152
def test_total_size(self):
157
164
class TestMixedTextStore(TestCaseInTempDir, TestStores):
159
166
def get_store(self, path=u'.', compressed=True):
160
t = LocalTransport(path)
167
t = transport.get_transport(path)
161
168
return TextStore(t, compressed=compressed)
163
170
def test_get_mixed(self):
357
364
my_store.get('missing', 'sig').read())
359
366
def test___iter__no_suffix(self):
360
my_store = TextStore(MemoryTransport(), False, compressed=False)
367
my_store = TextStore(MemoryTransport(),
368
prefixed=False, compressed=False)
361
369
stream = StringIO("content")
362
370
my_store.add(stream, "foo")
363
371
self.assertEqual(set(['foo']),
383
391
def test_copy_suffixes(self):
384
392
from_store = self.get_populated_store()
385
to_store = TextStore(MemoryTransport(), True, compressed=True)
393
to_store = TextStore(MemoryTransport(),
394
prefixed=True, compressed=True)
386
395
to_store.register_suffix('sig')
387
copy_all(from_store, to_store)
396
to_store.copy_all_ids(from_store)
388
397
self.assertEqual(1, len(to_store))
389
398
self.assertEqual(set(['foo']), set(to_store.__iter__()))
390
399
self.assertEqual('content', to_store.get('foo').read())
394
403
def test_relpath_escaped(self):
395
404
my_store = store.TransportStore(MemoryTransport())
396
405
self.assertEqual('%25', my_store._relpath('%'))
407
def test_escaped_uppercase(self):
408
"""Uppercase letters are escaped for safety on Windows"""
409
my_store = store.TransportStore(MemoryTransport(), escaped=True)
410
# a particularly perverse file-id! :-)
411
self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')
414
class TestVersionFileStore(TestCaseWithTransport):
417
super(TestVersionFileStore, self).setUp()
418
self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
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,
437
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)
443
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])