← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

Bugfix missing so tests pass.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
 
1
# Copyright (C) 2005 by Canonical Development Ltd
 
2
 
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
 
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
 
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Test Store implementations."""
 
18
 
 
19
from cStringIO import StringIO
 
20
import os
 
21
 
 
22
from bzrlib.errors import BzrError, UnlistableStore
 
23
from bzrlib.store import copy_all
 
24
from bzrlib.transport.local import LocalTransport
 
25
from bzrlib.transport import NoSuchFile
 
26
from bzrlib.store.compressed_text import CompressedTextStore
 
27
from bzrlib.store.text import TextStore
 
28
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
29
import bzrlib.store as store
 
30
import bzrlib.transport as transport
 
31
from bzrlib.transport.memory import MemoryTransport
 
32
 
 
33
 
 
34
class TestStores(object):
 
35
    """Mixin template class that provides some common tests for stores"""
 
36
 
 
37
    def check_content(self, store, fileid, value):
 
38
        f = store.get(fileid)
 
39
        self.assertEqual(f.read(), value)
 
40
 
 
41
    def fill_store(self, store):
 
42
        store.add(StringIO('hello'), 'a')
 
43
        store.add(StringIO('other'), 'b')
 
44
        store.add(StringIO('something'), 'c')
 
45
        store.add(StringIO('goodbye'), '123123')
 
46
 
 
47
    def test_copy_all(self):
 
48
        """Test copying"""
 
49
        os.mkdir('a')
 
50
        store_a = self.get_store('a')
 
51
        store_a.add('foo', '1')
 
52
        os.mkdir('b')
 
53
        store_b = self.get_store('b')
 
54
        copy_all(store_a, store_b)
 
55
        self.assertEqual(store_a.get('1').read(), 'foo')
 
56
        self.assertEqual(store_b.get('1').read(), 'foo')
 
57
        # TODO: Switch the exception form UnlistableStore to
 
58
        #       or make Stores throw UnlistableStore if their
 
59
        #       Transport doesn't support listing
 
60
        # store_c = RemoteStore('http://example.com/')
 
61
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
62
 
 
63
    def test_get(self):
 
64
        store = self.get_store()
 
65
        self.fill_store(store)
 
66
    
 
67
        self.check_content(store, 'a', 'hello')
 
68
        self.check_content(store, 'b', 'other')
 
69
        self.check_content(store, 'c', 'something')
 
70
    
 
71
        # Make sure that requesting a non-existing file fails
 
72
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
 
73
 
 
74
    def test_multiple_add(self):
 
75
        """Multiple add with same ID should raise a BzrError"""
 
76
        store = self.get_store()
 
77
        self.fill_store(store)
 
78
        self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
79
 
 
80
 
 
81
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
 
82
 
 
83
    def get_store(self, path='.'):
 
84
        t = LocalTransport(path)
 
85
        return CompressedTextStore(t)
 
86
 
 
87
    def test_total_size(self):
 
88
        store = self.get_store('.')
 
89
        store.register_suffix('dsc')
 
90
        store.add(StringIO('goodbye'), '123123')
 
91
        store.add(StringIO('goodbye2'), '123123', 'dsc')
 
92
        # these get gzipped - content should be stable
 
93
        self.assertEqual(store.total_size(), (2, 55))
 
94
        
 
95
    def test__relpath_suffixed(self):
 
96
        my_store = CompressedTextStore(MockTransport(), True)
 
97
        my_store.register_suffix('dsc')
 
98
        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))
 
99
 
 
100
 
 
101
class TestMemoryStore(TestCase):
 
102
    
 
103
    def get_store(self):
 
104
        return store.ImmutableMemoryStore()
 
105
    
 
106
    def test_imports(self):
 
107
        from bzrlib.store import ImmutableMemoryStore
 
108
 
 
109
    def test_add_and_retrieve(self):
 
110
        store = self.get_store()
 
111
        store.add(StringIO('hello'), 'aa')
 
112
        self.assertNotEqual(store.get('aa'), None)
 
113
        self.assertEqual(store.get('aa').read(), 'hello')
 
114
        store.add(StringIO('hello world'), 'bb')
 
115
        self.assertNotEqual(store.get('bb'), None)
 
116
        self.assertEqual(store.get('bb').read(), 'hello world')
 
117
 
 
118
    def test_missing_is_absent(self):
 
119
        store = self.get_store()
 
120
        self.failIf('aa' in store)
 
121
 
 
122
    def test_adding_fails_when_present(self):
 
123
        my_store = self.get_store()
 
124
        my_store.add(StringIO('hello'), 'aa')
 
125
        self.assertRaises(BzrError,
 
126
                          my_store.add, StringIO('hello'), 'aa')
 
127
 
 
128
    def test_total_size(self):
 
129
        store = self.get_store()
 
130
        store.add(StringIO('goodbye'), '123123')
 
131
        store.add(StringIO('goodbye2'), '123123.dsc')
 
132
        self.assertEqual(store.total_size(), (2, 15))
 
133
        # TODO: Switch the exception form UnlistableStore to
 
134
        #       or make Stores throw UnlistableStore if their
 
135
        #       Transport doesn't support listing
 
136
        # store_c = RemoteStore('http://example.com/')
 
137
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
138
 
 
139
 
 
140
class TestTextStore(TestCaseInTempDir, TestStores):
 
141
 
 
142
    def get_store(self, path='.'):
 
143
        t = LocalTransport(path)
 
144
        return TextStore(t)
 
145
 
 
146
    def test_total_size(self):
 
147
        store = self.get_store()
 
148
        store.add(StringIO('goodbye'), '123123')
 
149
        store.add(StringIO('goodbye2'), '123123.dsc')
 
150
        self.assertEqual(store.total_size(), (2, 15))
 
151
        # TODO: Switch the exception form UnlistableStore to
 
152
        #       or make Stores throw UnlistableStore if their
 
153
        #       Transport doesn't support listing
 
154
        # store_c = RemoteStore('http://example.com/')
 
155
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
156
 
 
157
 
 
158
class MockTransport(transport.Transport):
 
159
    """A fake transport for testing with."""
 
160
 
 
161
    def has(self, filename):
 
162
        return False
 
163
 
 
164
    def __init__(self, url=None):
 
165
        if url is None:
 
166
            url = "http://example.com"
 
167
        super(MockTransport, self).__init__(url)
 
168
 
 
169
    def mkdir(self, filename):
 
170
        return
 
171
 
 
172
 
 
173
class InstrumentedTransportStore(store.TransportStore):
 
174
    """An instrumented TransportStore.
 
175
 
 
176
    Here we replace template method worker methods with calls that record the
 
177
    expected results.
 
178
    """
 
179
 
 
180
    def _add(self, filename, file):
 
181
        self._calls.append(("_add", filename, file))
 
182
 
 
183
    def __init__(self, transport, prefixed=False):
 
184
        super(InstrumentedTransportStore, self).__init__(transport, prefixed)
 
185
        self._calls = []
 
186
 
 
187
 
 
188
class TestInstrumentedTransportStore(TestCase):
 
189
 
 
190
    def test__add_records(self):
 
191
        my_store = InstrumentedTransportStore(MockTransport())
 
192
        my_store._add("filename", "file")
 
193
        self.assertEqual([("_add", "filename", "file")], my_store._calls)
 
194
 
 
195
 
 
196
class TestMockTransport(TestCase):
 
197
 
 
198
    def test_isinstance(self):
 
199
        self.failUnless(isinstance(MockTransport(), transport.Transport))
 
200
 
 
201
    def test_has(self):
 
202
        self.assertEqual(False, MockTransport().has('foo'))
 
203
 
 
204
    def test_mkdir(self):
 
205
        MockTransport().mkdir('45')
 
206
 
 
207
 
 
208
class TestTransportStore(TestCase):
 
209
    
 
210
    def test__relpath_invalid(self):
 
211
        my_store = store.TransportStore(MockTransport())
 
212
        self.assertRaises(ValueError, my_store._relpath, '/foo')
 
213
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
 
214
 
 
215
    def test_register_invalid_suffixes(self):
 
216
        my_store = store.TransportStore(MockTransport())
 
217
        self.assertRaises(ValueError, my_store.register_suffix, '/')
 
218
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
 
219
 
 
220
    def test__relpath_unregister_suffixes(self):
 
221
        my_store = store.TransportStore(MockTransport())
 
222
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
 
223
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
 
224
 
 
225
    def test__relpath_simple(self):
 
226
        my_store = store.TransportStore(MockTransport())
 
227
        self.assertEqual("foo", my_store._relpath('foo'))
 
228
 
 
229
    def test__relpath_prefixed(self):
 
230
        my_store = store.TransportStore(MockTransport(), True)
 
231
        self.assertEqual('45/foo', my_store._relpath('foo'))
 
232
 
 
233
    def test__relpath_simple_suffixed(self):
 
234
        my_store = store.TransportStore(MockTransport())
 
235
        my_store.register_suffix('gz')
 
236
        my_store.register_suffix('bar')
 
237
        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))
 
238
        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))
 
239
 
 
240
    def test__relpath_prefixed_suffixed(self):
 
241
        my_store = store.TransportStore(MockTransport(), True)
 
242
        my_store.register_suffix('gz')
 
243
        my_store.register_suffix('bar')
 
244
        self.assertEqual('45/foo.gz', my_store._relpath('foo', ['gz']))
 
245
        self.assertEqual('45/foo.gz.bar',
 
246
                         my_store._relpath('foo', ['gz', 'bar']))
 
247
 
 
248
    def test_add_simple(self):
 
249
        stream = StringIO("content")
 
250
        my_store = InstrumentedTransportStore(MockTransport())
 
251
        my_store.add(stream, "foo")
 
252
        self.assertEqual([("_add", "foo", stream)], my_store._calls)
 
253
 
 
254
    def test_add_prefixed(self):
 
255
        stream = StringIO("content")
 
256
        my_store = InstrumentedTransportStore(MockTransport(), True)
 
257
        my_store.add(stream, "foo")
 
258
        self.assertEqual([("_add", "45/foo", stream)], my_store._calls)
 
259
 
 
260
    def test_add_simple_suffixed(self):
 
261
        stream = StringIO("content")
 
262
        my_store = InstrumentedTransportStore(MockTransport())
 
263
        my_store.register_suffix('dsc')
 
264
        my_store.add(stream, "foo", 'dsc')
 
265
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
 
266
        
 
267
    def test_add_simple_suffixed(self):
 
268
        stream = StringIO("content")
 
269
        my_store = InstrumentedTransportStore(MockTransport(), True)
 
270
        my_store.register_suffix('dsc')
 
271
        my_store.add(stream, "foo", 'dsc')
 
272
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
 
273
 
 
274
    def get_populated_store(self, prefixed=False, store_class=TextStore):
 
275
        my_store = store_class(MemoryTransport(), prefixed)
 
276
        my_store.register_suffix('sig')
 
277
        stream = StringIO("signature")
 
278
        my_store.add(stream, "foo", 'sig')
 
279
        stream = StringIO("content")
 
280
        my_store.add(stream, "foo")
 
281
        stream = StringIO("signature for missing base")
 
282
        my_store.add(stream, "missing", 'sig')
 
283
        return my_store
 
284
        
 
285
    def test_has_simple(self):
 
286
        my_store = self.get_populated_store()
 
287
        self.assertEqual(True, my_store.has_id('foo'))
 
288
        my_store = self.get_populated_store(True)
 
289
        self.assertEqual(True, my_store.has_id('foo'))
 
290
 
 
291
    def test_has_suffixed(self):
 
292
        my_store = self.get_populated_store()
 
293
        self.assertEqual(True, my_store.has_id('foo', 'sig'))
 
294
        my_store = self.get_populated_store(True)
 
295
        self.assertEqual(True, my_store.has_id('foo', 'sig'))
 
296
 
 
297
    def test_has_suffixed_no_base(self):
 
298
        my_store = self.get_populated_store()
 
299
        self.assertEqual(False, my_store.has_id('missing'))
 
300
        my_store = self.get_populated_store(True)
 
301
        self.assertEqual(False, my_store.has_id('missing'))
 
302
 
 
303
    def test_get_simple(self):
 
304
        my_store = self.get_populated_store()
 
305
        self.assertEqual('content', my_store.get('foo').read())
 
306
        my_store = self.get_populated_store(True)
 
307
        self.assertEqual('content', my_store.get('foo').read())
 
308
 
 
309
    def test_get_suffixed(self):
 
310
        my_store = self.get_populated_store()
 
311
        self.assertEqual('signature', my_store.get('foo', 'sig').read())
 
312
        my_store = self.get_populated_store(True)
 
313
        self.assertEqual('signature', my_store.get('foo', 'sig').read())
 
314
 
 
315
    def test_get_suffixed_no_base(self):
 
316
        my_store = self.get_populated_store()
 
317
        self.assertEqual('signature for missing base',
 
318
                         my_store.get('missing', 'sig').read())
 
319
        my_store = self.get_populated_store(True)
 
320
        self.assertEqual('signature for missing base',
 
321
                         my_store.get('missing', 'sig').read())
 
322
 
 
323
    def test___iter__no_suffix(self):
 
324
        my_store = TextStore(MemoryTransport(), False)
 
325
        stream = StringIO("content")
 
326
        my_store.add(stream, "foo")
 
327
        self.assertEqual(set(['foo']),
 
328
                         set(my_store.__iter__()))
 
329
 
 
330
    def test___iter__(self):
 
331
        self.assertEqual(set(['foo']),
 
332
                         set(self.get_populated_store().__iter__()))
 
333
        self.assertEqual(set(['foo']),
 
334
                         set(self.get_populated_store(True).__iter__()))
 
335
 
 
336
    def test___iter__compressed(self):
 
337
        self.assertEqual(set(['foo']),
 
338
                         set(self.get_populated_store(
 
339
                             store_class=CompressedTextStore).__iter__()))
 
340
        self.assertEqual(set(['foo']),
 
341
                         set(self.get_populated_store(
 
342
                             True, CompressedTextStore).__iter__()))
 
343
 
 
344
    def test___len__(self):
 
345
        self.assertEqual(1, len(self.get_populated_store()))
 
346
 
 
347
    def test_copy_suffixes(self):
 
348
        from_store = self.get_populated_store()
 
349
        to_store = CompressedTextStore(MemoryTransport(), True)
 
350
        to_store.register_suffix('sig')
 
351
        copy_all(from_store, to_store)
 
352
        self.assertEqual(1, len(to_store))
 
353
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
 
354
        self.assertEqual('content', to_store.get('foo').read())
 
355
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
 
356
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
 
357
 
 
358
    def test_relpath_escaped(self):
 
359
        my_store = store.TransportStore(MemoryTransport())
 
360
        self.assertEqual('%25', my_store._relpath('%'))