/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Robert Collins
  • Date: 2010-05-06 23:41:35 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506234135-yivbzczw1sejxnxc
Lock methods on ``Tree``, ``Branch`` and ``Repository`` are now
expected to return an object which can be used to unlock them. This reduces
duplicate code when using cleanups. The previous 'tokens's returned by
``Branch.lock_write`` and ``Repository.lock_write`` are now attributes
on the result of the lock_write. ``repository.RepositoryWriteLockResult``
and ``branch.BranchWriteLockResult`` document this. (Robert Collins)

``log._get_info_for_log_files`` now takes an add_cleanup callable.
(Robert Collins)

Show diffs side-by-side

added added

removed removed

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