24
24
from bzrlib.transport import NoSuchFile
25
25
from bzrlib.store.compressed_text import CompressedTextStore
26
26
from bzrlib.store.text import TextStore
27
from bzrlib.selftest import TestCaseInTempDir
27
from bzrlib.selftest import TestCase, TestCaseInTempDir
28
28
from bzrlib.errors import BzrError, UnlistableStore
31
32
def fill_store(store):
92
93
check_equals(tester, store, ['b', 'd', 'c'], ['other', None, 'something'],
93
94
permit_failure=True)
95
97
def get_compressed_store(path='.'):
96
98
t = LocalTransport(path)
97
99
return CompressedTextStore(t)
99
102
def get_text_store(path='.'):
100
103
t = LocalTransport(path)
101
104
return TextStore(t)
103
107
class TestCompressedTextStore(TestCaseInTempDir):
104
109
def test_multiple_add(self):
105
110
"""Multiple add with same ID should raise a BzrError"""
106
111
store = get_compressed_store()
115
120
test_ignore_get(self, store)
123
def test_total_size(self):
124
store = get_compressed_store('.')
125
store.add(StringIO('goodbye'), '123123')
126
store.add(StringIO('goodbye2'), '123123.dsc')
127
# these get gzipped - content should be stable
128
self.assertEqual(store.total_size(), (2, 55))
118
130
def test_copy_all(self):
119
131
"""Test copying"""
125
137
copy_all(store_a, store_b)
126
138
self.assertEqual(store_a['1'].read(), 'foo')
127
139
self.assertEqual(store_b['1'].read(), 'foo')
141
class TestMemoryStore(TestCase):
144
return bzrlib.store.ImmutableMemoryStore()
146
def test_imports(self):
147
from bzrlib.store import ImmutableMemoryStore
149
def test_add_and_retrieve(self):
150
store = self.get_store()
151
store.add(StringIO('hello'), 'aa')
152
self.assertNotEqual(store['aa'], None)
153
self.assertEqual(store['aa'].read(), 'hello')
154
store.add(StringIO('hello world'), 'bb')
155
self.assertNotEqual(store['bb'], None)
156
self.assertEqual(store['bb'].read(), 'hello world')
158
def test_missing_is_absent(self):
159
store = self.get_store()
160
self.failIf('aa' in store)
162
def test_adding_fails_when_present(self):
163
store = self.get_store()
164
store.add(StringIO('hello'), 'aa')
165
self.assertRaises(bzrlib.store.StoreError,
166
store.add, StringIO('hello'), 'aa')
168
def test_total_size(self):
169
store = self.get_store()
170
store.add(StringIO('goodbye'), '123123')
171
store.add(StringIO('goodbye2'), '123123.dsc')
172
self.assertEqual(store.total_size(), (2, 15))
128
173
# TODO: Switch the exception form UnlistableStore to
129
174
# or make Stores throw UnlistableStore if their
130
175
# Transport doesn't support listing
161
206
# Transport doesn't support listing
162
207
# store_c = RemoteStore('http://example.com/')
163
208
# self.assertRaises(UnlistableStore, copy_all, store_c, store_b)