← Back to branch summary

~bzr/ubuntu/lucid/bzr/beta-ppa

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Martin Pool
  • Date: 2010-07-02 07:29:40 UTC
  • mfrom: (129.1.7 packaging-karmic)
  • Revision ID: mbp@sourcefrog.net-20100702072940-hpzq5elg8wjve8rh
http://bugs.debian.org/452024
http://bugs.debian.org/464688
http://bugs.debian.org/523595
http://bugs.debian.org/526233
http://bugs.debian.org/537708
http://bugs.debian.org/555336
http://bugs.debian.org/555343
https://launchpad.net/bugs/249452
https://launchpad.net/bugs/385074
* PPA rebuild.
* PPA rebuild for Karmic.
* PPA rebuild for Jaunty.
* PPA rebuild for Hardy.
* From postinst, actually remove the example bash completion scripts.
  (LP: #249452)
* New upstream release.
* New upstream release.
* New upstream release.
* Revert change to Build-depends: Dapper does not have python-central.
  Should be python-support..
* Target ppa..
* Target ppa..
* Target ppa..
* Target ppa..
* New upstream release.
* Switch to dpkg-source 3.0 (quilt) format.
* Bump standards version to 3.8.4.
* Remove embedded copy of python-configobj. Closes: #555336
* Remove embedded copy of python-elementtree. Closes: #555343
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* Change section from 'Devel' to 'Vcs'..
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* debian/control: Fix obsolete-relation-form-in-source
  lintian warning. 
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Split out docs into bzr-doc package.
* New upstream release.
* Added John Francesco Ferlito to Uploaders.
* Fix install path to quick-reference guide
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to doc paths changing
* New upstream release.
* Fix FTBFS due to path changes, again, again.
* Fix FTBFS due to path changes, again.
* Fix FTBFS due to path changes.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Bump standards version to 3.8.3.
* Remove unused patch system.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix copy and paste tab error in .install file
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
 + Fixes compatibility with Python 2.4. Closes: #537708
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream version.
* Bump standards version to 3.8.2.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add python-pyrex to build-deps to ensure C extensions are always build.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* Split documentation into bzr-doc package. ((LP: #385074)
* Multiple packaging changes to make us more linitan clean.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Fix API compatibility version. (Closes: #526233)
* New upstream release.
  + Fixes default format for upgrade command. (Closes: #464688)
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add missing dependency on zlib development library. (Closes:
  #523595)
* Add zlib build-depends.
* Add zlib build-depends.
* Add zlib build-depends.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Move to section vcs.
* Bump standards version to 3.8.1.
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* New upstream release.
* Remove temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* Add temporary patch for missing .c files from distribution
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Recommend ca-certificates. (Closes: #452024)
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Update watch file. bazaar now uses launchpad to host its sources.
* Remove patch for inventory root revision copy, applied upstream.
* New upstream release.
* New upstream release.
* New upstream release
* Force removal of files installed in error to /etc/bash_completion.d/
  (LP: #249452)
* New upstream release.
* New upstream release
* New upstream release.
* Bump standards version.
* Include patch for inventory root revision copy, required for bzr-svn.
* New upstream release.
* Remove unused lintian overrides.
* Correct the package version not to be native.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* New upstream release.
* Final 1.5 release.
* New upstream release.
* New upstream release.
* New upstream release.
* Add myself as a co-maintainer.
* Add a Dm-Upload-Allowed: yes header.
* New upstream bugfix release.
* New upstream release.
* Final 1.3 release.
* New upstream release.
* First release candidate of the upcoming 1.3 release.
* Rebuild to fix the problem caused by a build with a broken python-central.
* New upstream release.
* Rebuild for dapper PPA.
* Apply Lamont's patches to fix build-dependencies on dapper.
  (See: https://bugs.launchpad.net/bzr/+bug/189915)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_store.py
 
1
# Copyright (C) 2005, 2007 Canonical 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Test Store implementations."""
 
18
 
 
19
from cStringIO import StringIO
 
20
import os
 
21
import gzip
 
22
 
 
23
import bzrlib.errors as errors
 
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
 
25
from bzrlib.transport.local import LocalTransport
 
26
from bzrlib.store.text import TextStore
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
 
28
import bzrlib.store as store
 
29
import bzrlib.store.versioned
 
30
import bzrlib.transactions as transactions
 
31
import bzrlib.transport as transport
 
32
from bzrlib.transport.memory import MemoryTransport
 
33
 
 
34
 
 
35
class TestStores(object):
 
36
    """Mixin template class that provides some common tests for stores"""
 
37
 
 
38
    def check_content(self, store, fileid, value):
 
39
        f = store.get(fileid)
 
40
        self.assertEqual(f.read(), value)
 
41
 
 
42
    def fill_store(self, store):
 
43
        store.add(StringIO('hello'), 'a')
 
44
        store.add(StringIO('other'), 'b')
 
45
        store.add(StringIO('something'), 'c')
 
46
        store.add(StringIO('goodbye'), '123123')
 
47
 
 
48
    def test_copy_all(self):
 
49
        """Test copying"""
 
50
        os.mkdir('a')
 
51
        store_a = self.get_store('a')
 
52
        store_a.add(StringIO('foo'), '1')
 
53
        os.mkdir('b')
 
54
        store_b = self.get_store('b')
 
55
        store_b.copy_all_ids(store_a)
 
56
        self.assertEqual(store_a.get('1').read(), 'foo')
 
57
        self.assertEqual(store_b.get('1').read(), 'foo')
 
58
        # TODO: Switch the exception form UnlistableStore to
 
59
        #       or make Stores throw UnlistableStore if their
 
60
        #       Transport doesn't support listing
 
61
        # store_c = RemoteStore('http://example.com/')
 
62
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
63
 
 
64
    def test_get(self):
 
65
        store = self.get_store()
 
66
        self.fill_store(store)
 
67
 
 
68
        self.check_content(store, 'a', 'hello')
 
69
        self.check_content(store, 'b', 'other')
 
70
        self.check_content(store, 'c', 'something')
 
71
 
 
72
        # Make sure that requesting a non-existing file fails
 
73
        self.assertRaises(KeyError, self.check_content, store, 'd', None)
 
74
 
 
75
    def test_multiple_add(self):
 
76
        """Multiple add with same ID should raise a BzrError"""
 
77
        store = self.get_store()
 
78
        self.fill_store(store)
 
79
        self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
80
 
 
81
 
 
82
class TestCompressedTextStore(TestCaseInTempDir, TestStores):
 
83
 
 
84
    def get_store(self, path=u'.'):
 
85
        t = transport.get_transport(path)
 
86
        return TextStore(t, compressed=True)
 
87
 
 
88
    def test_total_size(self):
 
89
        store = self.get_store(u'.')
 
90
        store.register_suffix('dsc')
 
91
        store.add(StringIO('goodbye'), '123123')
 
92
        store.add(StringIO('goodbye2'), '123123', 'dsc')
 
93
        # these get gzipped - content should be stable
 
94
        self.assertEqual(store.total_size(), (2, 55))
 
95
 
 
96
    def test__relpath_suffixed(self):
 
97
        my_store = TextStore(MockTransport(),
 
98
                             prefixed=True, compressed=True)
 
99
        my_store.register_suffix('dsc')
 
100
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
 
101
 
 
102
 
 
103
class TestMemoryStore(TestCase):
 
104
 
 
105
    def get_store(self):
 
106
        return TextStore(MemoryTransport())
 
107
 
 
108
    def test_add_and_retrieve(self):
 
109
        store = self.get_store()
 
110
        store.add(StringIO('hello'), 'aa')
 
111
        self.assertNotEqual(store.get('aa'), None)
 
112
        self.assertEqual(store.get('aa').read(), 'hello')
 
113
        store.add(StringIO('hello world'), 'bb')
 
114
        self.assertNotEqual(store.get('bb'), None)
 
115
        self.assertEqual(store.get('bb').read(), 'hello world')
 
116
 
 
117
    def test_missing_is_absent(self):
 
118
        store = self.get_store()
 
119
        self.failIf('aa' in store)
 
120
 
 
121
    def test_adding_fails_when_present(self):
 
122
        my_store = self.get_store()
 
123
        my_store.add(StringIO('hello'), 'aa')
 
124
        self.assertRaises(BzrError,
 
125
                          my_store.add, StringIO('hello'), 'aa')
 
126
 
 
127
    def test_total_size(self):
 
128
        store = self.get_store()
 
129
        store.add(StringIO('goodbye'), '123123')
 
130
        store.add(StringIO('goodbye2'), '123123.dsc')
 
131
        self.assertEqual(store.total_size(), (2, 15))
 
132
        # TODO: Switch the exception form UnlistableStore to
 
133
        #       or make Stores throw UnlistableStore if their
 
134
        #       Transport doesn't support listing
 
135
        # store_c = RemoteStore('http://example.com/')
 
136
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
137
 
 
138
 
 
139
class TestTextStore(TestCaseInTempDir, TestStores):
 
140
 
 
141
    def get_store(self, path=u'.'):
 
142
        t = transport.get_transport(path)
 
143
        return TextStore(t, compressed=False)
 
144
 
 
145
    def test_total_size(self):
 
146
        store = self.get_store()
 
147
        store.add(StringIO('goodbye'), '123123')
 
148
        store.add(StringIO('goodbye2'), '123123.dsc')
 
149
        self.assertEqual(store.total_size(), (2, 15))
 
150
        # TODO: Switch the exception form UnlistableStore to
 
151
        #       or make Stores throw UnlistableStore if their
 
152
        #       Transport doesn't support listing
 
153
        # store_c = RemoteStore('http://example.com/')
 
154
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
155
 
 
156
 
 
157
class TestMixedTextStore(TestCaseInTempDir, TestStores):
 
158
 
 
159
    def get_store(self, path=u'.', compressed=True):
 
160
        t = transport.get_transport(path)
 
161
        return TextStore(t, compressed=compressed)
 
162
 
 
163
    def test_get_mixed(self):
 
164
        cs = self.get_store(u'.', compressed=True)
 
165
        s = self.get_store(u'.', compressed=False)
 
166
        cs.add(StringIO('hello there'), 'a')
 
167
 
 
168
        self.failUnlessExists('a.gz')
 
169
        self.failIf(os.path.lexists('a'))
 
170
 
 
171
        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')
 
172
 
 
173
        self.assertEquals(cs.has_id('a'), True)
 
174
        self.assertEquals(s.has_id('a'), True)
 
175
        self.assertEquals(cs.get('a').read(), 'hello there')
 
176
        self.assertEquals(s.get('a').read(), 'hello there')
 
177
 
 
178
        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')
 
179
 
 
180
        s.add(StringIO('goodbye'), 'b')
 
181
        self.failUnlessExists('b')
 
182
        self.failIf(os.path.lexists('b.gz'))
 
183
        self.assertEquals(open('b').read(), 'goodbye')
 
184
 
 
185
        self.assertEquals(cs.has_id('b'), True)
 
186
        self.assertEquals(s.has_id('b'), True)
 
187
        self.assertEquals(cs.get('b').read(), 'goodbye')
 
188
        self.assertEquals(s.get('b').read(), 'goodbye')
 
189
 
 
190
        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')
 
191
 
 
192
class MockTransport(transport.Transport):
 
193
    """A fake transport for testing with."""
 
194
 
 
195
    def has(self, filename):
 
196
        return False
 
197
 
 
198
    def __init__(self, url=None):
 
199
        if url is None:
 
200
            url = "http://example.com"
 
201
        super(MockTransport, self).__init__(url)
 
202
 
 
203
    def mkdir(self, filename):
 
204
        return
 
205
 
 
206
 
 
207
class InstrumentedTransportStore(store.TransportStore):
 
208
    """An instrumented TransportStore.
 
209
 
 
210
    Here we replace template method worker methods with calls that record the
 
211
    expected results.
 
212
    """
 
213
 
 
214
    def _add(self, filename, file):
 
215
        self._calls.append(("_add", filename, file))
 
216
 
 
217
    def __init__(self, transport, prefixed=False):
 
218
        super(InstrumentedTransportStore, self).__init__(transport, prefixed)
 
219
        self._calls = []
 
220
 
 
221
 
 
222
class TestInstrumentedTransportStore(TestCase):
 
223
 
 
224
    def test__add_records(self):
 
225
        my_store = InstrumentedTransportStore(MockTransport())
 
226
        my_store._add("filename", "file")
 
227
        self.assertEqual([("_add", "filename", "file")], my_store._calls)
 
228
 
 
229
 
 
230
class TestMockTransport(TestCase):
 
231
 
 
232
    def test_isinstance(self):
 
233
        self.failUnless(isinstance(MockTransport(), transport.Transport))
 
234
 
 
235
    def test_has(self):
 
236
        self.assertEqual(False, MockTransport().has('foo'))
 
237
 
 
238
    def test_mkdir(self):
 
239
        MockTransport().mkdir('45')
 
240
 
 
241
 
 
242
class TestTransportStore(TestCase):
 
243
 
 
244
    def test__relpath_invalid(self):
 
245
        my_store = store.TransportStore(MockTransport())
 
246
        self.assertRaises(ValueError, my_store._relpath, '/foo')
 
247
        self.assertRaises(ValueError, my_store._relpath, 'foo/')
 
248
 
 
249
    def test_register_invalid_suffixes(self):
 
250
        my_store = store.TransportStore(MockTransport())
 
251
        self.assertRaises(ValueError, my_store.register_suffix, '/')
 
252
        self.assertRaises(ValueError, my_store.register_suffix, '.gz/bar')
 
253
 
 
254
    def test__relpath_unregister_suffixes(self):
 
255
        my_store = store.TransportStore(MockTransport())
 
256
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['gz'])
 
257
        self.assertRaises(ValueError, my_store._relpath, 'foo', ['dsc', 'gz'])
 
258
 
 
259
    def test__relpath_simple(self):
 
260
        my_store = store.TransportStore(MockTransport())
 
261
        self.assertEqual("foo", my_store._relpath('foo'))
 
262
 
 
263
    def test__relpath_prefixed(self):
 
264
        my_store = store.TransportStore(MockTransport(), True)
 
265
        self.assertEqual('45/foo', my_store._relpath('foo'))
 
266
 
 
267
    def test__relpath_simple_suffixed(self):
 
268
        my_store = store.TransportStore(MockTransport())
 
269
        my_store.register_suffix('bar')
 
270
        my_store.register_suffix('baz')
 
271
        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))
 
272
        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))
 
273
 
 
274
    def test__relpath_prefixed_suffixed(self):
 
275
        my_store = store.TransportStore(MockTransport(), True)
 
276
        my_store.register_suffix('bar')
 
277
        my_store.register_suffix('baz')
 
278
        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))
 
279
        self.assertEqual('45/foo.bar.baz',
 
280
                         my_store._relpath('foo', ['bar', 'baz']))
 
281
 
 
282
    def test_add_simple(self):
 
283
        stream = StringIO("content")
 
284
        my_store = InstrumentedTransportStore(MockTransport())
 
285
        my_store.add(stream, "foo")
 
286
        self.assertEqual([("_add", "foo", stream)], my_store._calls)
 
287
 
 
288
    def test_add_prefixed(self):
 
289
        stream = StringIO("content")
 
290
        my_store = InstrumentedTransportStore(MockTransport(), True)
 
291
        my_store.add(stream, "foo")
 
292
        self.assertEqual([("_add", "45/foo", stream)], my_store._calls)
 
293
 
 
294
    def test_add_simple_suffixed(self):
 
295
        stream = StringIO("content")
 
296
        my_store = InstrumentedTransportStore(MockTransport())
 
297
        my_store.register_suffix('dsc')
 
298
        my_store.add(stream, "foo", 'dsc')
 
299
        self.assertEqual([("_add", "foo.dsc", stream)], my_store._calls)
 
300
 
 
301
    def test_add_simple_suffixed(self):
 
302
        stream = StringIO("content")
 
303
        my_store = InstrumentedTransportStore(MockTransport(), True)
 
304
        my_store.register_suffix('dsc')
 
305
        my_store.add(stream, "foo", 'dsc')
 
306
        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)
 
307
 
 
308
    def get_populated_store(self, prefixed=False,
 
309
            store_class=TextStore, compressed=False):
 
310
        my_store = store_class(MemoryTransport(), prefixed,
 
311
                               compressed=compressed)
 
312
        my_store.register_suffix('sig')
 
313
        stream = StringIO("signature")
 
314
        my_store.add(stream, "foo", 'sig')
 
315
        stream = StringIO("content")
 
316
        my_store.add(stream, "foo")
 
317
        stream = StringIO("signature for missing base")
 
318
        my_store.add(stream, "missing", 'sig')
 
319
        return my_store
 
320
 
 
321
    def test_has_simple(self):
 
322
        my_store = self.get_populated_store()
 
323
        self.assertEqual(True, my_store.has_id('foo'))
 
324
        my_store = self.get_populated_store(True)
 
325
        self.assertEqual(True, my_store.has_id('foo'))
 
326
 
 
327
    def test_has_suffixed(self):
 
328
        my_store = self.get_populated_store()
 
329
        self.assertEqual(True, my_store.has_id('foo', 'sig'))
 
330
        my_store = self.get_populated_store(True)
 
331
        self.assertEqual(True, my_store.has_id('foo', 'sig'))
 
332
 
 
333
    def test_has_suffixed_no_base(self):
 
334
        my_store = self.get_populated_store()
 
335
        self.assertEqual(False, my_store.has_id('missing'))
 
336
        my_store = self.get_populated_store(True)
 
337
        self.assertEqual(False, my_store.has_id('missing'))
 
338
 
 
339
    def test_get_simple(self):
 
340
        my_store = self.get_populated_store()
 
341
        self.assertEqual('content', my_store.get('foo').read())
 
342
        my_store = self.get_populated_store(True)
 
343
        self.assertEqual('content', my_store.get('foo').read())
 
344
 
 
345
    def test_get_suffixed(self):
 
346
        my_store = self.get_populated_store()
 
347
        self.assertEqual('signature', my_store.get('foo', 'sig').read())
 
348
        my_store = self.get_populated_store(True)
 
349
        self.assertEqual('signature', my_store.get('foo', 'sig').read())
 
350
 
 
351
    def test_get_suffixed_no_base(self):
 
352
        my_store = self.get_populated_store()
 
353
        self.assertEqual('signature for missing base',
 
354
                         my_store.get('missing', 'sig').read())
 
355
        my_store = self.get_populated_store(True)
 
356
        self.assertEqual('signature for missing base',
 
357
                         my_store.get('missing', 'sig').read())
 
358
 
 
359
    def test___iter__no_suffix(self):
 
360
        my_store = TextStore(MemoryTransport(),
 
361
                             prefixed=False, compressed=False)
 
362
        stream = StringIO("content")
 
363
        my_store.add(stream, "foo")
 
364
        self.assertEqual(set(['foo']),
 
365
                         set(my_store.__iter__()))
 
366
 
 
367
    def test___iter__(self):
 
368
        self.assertEqual(set(['foo']),
 
369
                         set(self.get_populated_store().__iter__()))
 
370
        self.assertEqual(set(['foo']),
 
371
                         set(self.get_populated_store(True).__iter__()))
 
372
 
 
373
    def test___iter__compressed(self):
 
374
        self.assertEqual(set(['foo']),
 
375
                         set(self.get_populated_store(
 
376
                             compressed=True).__iter__()))
 
377
        self.assertEqual(set(['foo']),
 
378
                         set(self.get_populated_store(
 
379
                             True, compressed=True).__iter__()))
 
380
 
 
381
    def test___len__(self):
 
382
        self.assertEqual(1, len(self.get_populated_store()))
 
383
 
 
384
    def test_copy_suffixes(self):
 
385
        from_store = self.get_populated_store()
 
386
        to_store = TextStore(MemoryTransport(),
 
387
                             prefixed=True, compressed=True)
 
388
        to_store.register_suffix('sig')
 
389
        to_store.copy_all_ids(from_store)
 
390
        self.assertEqual(1, len(to_store))
 
391
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
 
392
        self.assertEqual('content', to_store.get('foo').read())
 
393
        self.assertEqual('signature', to_store.get('foo', 'sig').read())
 
394
        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')
 
395
 
 
396
    def test_relpath_escaped(self):
 
397
        my_store = store.TransportStore(MemoryTransport())
 
398
        self.assertEqual('%25', my_store._relpath('%'))
 
399
 
 
400
    def test_escaped_uppercase(self):
 
401
        """Uppercase letters are escaped for safety on Windows"""
 
402
        my_store = store.TransportStore(MemoryTransport(), prefixed=True,
 
403
            escaped=True)
 
404
        # a particularly perverse file-id! :-)
 
405
        self.assertEquals(my_store._relpath('C:<>'), 'be/%2543%253a%253c%253e')
 
406
 
 
407
 
 
408
class TestVersionFileStore(TestCaseWithTransport):
 
409
 
 
410
    def get_scope(self):
 
411
        return self._transaction
 
412
 
 
413
    def setUp(self):
 
414
        super(TestVersionFileStore, self).setUp()
 
415
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
416
        self.vfstore.get_scope = self.get_scope
 
417
        self._transaction = None
 
418
 
 
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', [], [])
 
430
 
 
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', [], [])
 
438
 
 
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', [], [])
 
446
        del vf
 
447
        self._transaction.finish()
 
448
        self.assertEqual([' '], list(self.vfstore))