← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockable_files.py

  • Committer: Vincent Ladeuil
  • Date: 2017-01-17 13:48:10 UTC
  • mfrom: (6615.3.6 merges)
  • mto: This revision was merged to the branch mainline in revision 6620.
  • Revision ID: v.ladeuil+lp@free.fr-20170117134810-j9p3lidfy6pfyfsc
Merge 2.7, resolving conflicts

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
1
 
# Copyright (C) 2005, 2006, 2008 Canonical Ltd
 
1
# Copyright (C) 2005-2011 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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
from StringIO import StringIO
18
 
 
19
17
import bzrlib
20
18
from bzrlib import (
21
19
    errors,
22
20
    lockdir,
23
21
    osutils,
 
22
    transport,
24
23
    )
25
 
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
26
24
from bzrlib.lockable_files import LockableFiles, TransportLock
27
 
from bzrlib.symbol_versioning import (
28
 
    deprecated_in,
29
 
    )
30
25
from bzrlib.tests import (
31
26
    TestCaseInTempDir,
32
27
    TestNotApplicable,
37
32
                                 ReadOnlyTransaction,
38
33
                                 WriteTransaction,
39
34
                                 )
40
 
from bzrlib.transport import get_transport
41
35
 
42
36
 
43
37
# these tests are applied in each parameterized suite for LockableFiles
46
40
# so won't modernize them now. - mbp 20080430
47
41
class _TestLockableFiles_mixin(object):
48
42
 
49
 
    def test_read_write(self):
50
 
        self.assertRaises(NoSuchFile,
51
 
            self.applyDeprecated,
52
 
            deprecated_in((1, 5, 0)),
53
 
            self.lockable.get, 'foo')
54
 
        self.assertRaises(NoSuchFile,
55
 
            self.applyDeprecated,
56
 
            deprecated_in((1, 5, 0)),
57
 
            self.lockable.get_utf8, 'foo')
58
 
        self.lockable.lock_write()
59
 
        self.addCleanup(self.lockable.unlock)
60
 
        unicode_string = u'bar\u1234'
61
 
        self.assertEqual(4, len(unicode_string))
62
 
        byte_string = unicode_string.encode('utf-8')
63
 
        self.assertEqual(6, len(byte_string))
64
 
        self.assertRaises(UnicodeEncodeError,
65
 
            self.applyDeprecated,
66
 
            deprecated_in((1, 6, 0)),
67
 
            self.lockable.put, 'foo',
68
 
            StringIO(unicode_string))
69
 
        self.applyDeprecated(
70
 
            deprecated_in((1, 6, 0)),
71
 
            self.lockable.put,
72
 
            'foo', StringIO(byte_string))
73
 
        byte_stream = self.applyDeprecated(
74
 
            deprecated_in((1, 5, 0)),
75
 
            self.lockable.get,
76
 
            'foo')
77
 
        self.assertEqual(byte_string, byte_stream.read())
78
 
        unicode_stream = self.applyDeprecated(
79
 
            deprecated_in((1, 5, 0)),
80
 
            self.lockable.get_utf8,
81
 
            'foo')
82
 
        self.assertEqual(unicode_string,
83
 
            unicode_stream.read())
84
 
        self.assertRaises(BzrBadParameterNotString,
85
 
            self.applyDeprecated,
86
 
            deprecated_in((1, 6, 0)),
87
 
            self.lockable.put_utf8,
88
 
            'bar',
89
 
            StringIO(unicode_string))
90
 
        self.applyDeprecated(
91
 
            deprecated_in((1, 6, 0)),
92
 
            self.lockable.put_utf8,
93
 
            'bar',
94
 
            unicode_string)
95
 
        unicode_stream = self.applyDeprecated(
96
 
            deprecated_in((1, 5, 0)),
97
 
            self.lockable.get_utf8,
98
 
            'bar')
99
 
        self.assertEqual(unicode_string,
100
 
            unicode_stream.read())
101
 
        byte_stream = self.applyDeprecated(
102
 
            deprecated_in((1, 5, 0)),
103
 
            self.lockable.get,
104
 
            'bar')
105
 
        self.assertEqual(byte_string, byte_stream.read())
106
 
        self.applyDeprecated(
107
 
            deprecated_in((1, 6, 0)),
108
 
            self.lockable.put_bytes,
109
 
            'raw', 'raw\xffbytes')
110
 
        byte_stream = self.applyDeprecated(
111
 
            deprecated_in((1, 5, 0)),
112
 
            self.lockable.get,
113
 
            'raw')
114
 
        self.assertEqual('raw\xffbytes', byte_stream.read())
115
 
 
116
 
    def test_locks(self):
117
 
        self.lockable.lock_read()
118
 
        self.addCleanup(self.lockable.unlock)
119
 
        self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
120
 
                          StringIO('bar\u1234'))
121
 
 
122
43
    def test_transactions(self):
123
44
        self.assertIs(self.lockable.get_transaction().__class__,
124
45
                      PassThroughTransaction)
351
272
                                      _TestLockableFiles_mixin):
352
273
 
353
274
    def setUp(self):
354
 
        TestCaseInTempDir.setUp(self)
355
 
        transport = get_transport('.')
356
 
        transport.mkdir('.bzr')
357
 
        self.sub_transport = transport.clone('.bzr')
 
275
        super(TestLockableFiles_TransportLock, self).setUp()
 
276
        t = transport.get_transport_from_path('.')
 
277
        t.mkdir('.bzr')
 
278
        self.sub_transport = t.clone('.bzr')
358
279
        self.lockable = self.get_lockable()
359
280
        self.lockable.create_lock()
360
281
 
361
 
    def tearDown(self):
362
 
        super(TestLockableFiles_TransportLock, self).tearDown()
 
282
    def stop_server(self):
 
283
        super(TestLockableFiles_TransportLock, self).stop_server()
363
284
        # free the subtransport so that we do not get a 5 second
364
285
        # timeout due to the SFTP connection cache.
365
286
        try:
372
293
 
373
294
 
374
295
class TestLockableFiles_LockDir(TestCaseInTempDir,
375
 
                              _TestLockableFiles_mixin):
 
296
                                _TestLockableFiles_mixin):
376
297
    """LockableFile tests run with LockDir underneath"""
377
298
 
378
299
    def setUp(self):
379
 
        TestCaseInTempDir.setUp(self)
380
 
        self.transport = get_transport('.')
 
300
        super(TestLockableFiles_LockDir, self).setUp()
 
301
        self.transport = transport.get_transport_from_path('.')
381
302
        self.lockable = self.get_lockable()
382
303
        # the lock creation here sets mode - test_permissions on branch
383
304
        # tests that implicitly, but it might be a good idea to factor
408
329
 
409
330
 
410
331
class TestLockableFiles_RemoteLockDir(TestCaseWithSmartMedium,
411
 
                              _TestLockableFiles_mixin):
 
332
                                      _TestLockableFiles_mixin):
412
333
    """LockableFile tests run with RemoteLockDir on a branch."""
413
334
 
414
335
    def setUp(self):
415
 
        TestCaseWithSmartMedium.setUp(self)
 
336
        super(TestLockableFiles_RemoteLockDir, self).setUp()
416
337
        # can only get a RemoteLockDir with some RemoteObject...
417
338
        # use a branch as thats what we want. These mixin tests test the end
418
339
        # to end behaviour, so stubbing out the backend and simulating would
420
341
        # in test_remote and test_smart as usual.
421
342
        b = self.make_branch('foo')
422
343
        self.addCleanup(b.bzrdir.transport.disconnect)
423
 
        self.transport = get_transport('.')
 
344
        self.transport = transport.get_transport_from_path('.')
424
345
        self.lockable = self.get_lockable()
425
346
 
426
347
    def get_lockable(self):