← 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: Martin Pool
  • Date: 2008-06-11 02:36:40 UTC
  • mfrom: (3490 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3492.
  • Revision ID: mbp@sourcefrog.net-20080611023640-db0lqd75yueksdw7
Merge news

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
27
27
from bzrlib.symbol_versioning import (
28
28
    deprecated_in,
29
29
    )
30
 
from bzrlib.tests import TestCaseInTempDir
 
30
from bzrlib.tests import (
 
31
    TestCaseInTempDir,
 
32
    TestNotApplicable,
 
33
    )
31
34
from bzrlib.tests.test_smart import TestCaseWithSmartMedium
32
35
from bzrlib.tests.test_transactions import DummyWeave
33
36
from bzrlib.transactions import (PassThroughTransaction,
44
47
class _TestLockableFiles_mixin(object):
45
48
 
46
49
    def test_read_write(self):
47
 
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
 
50
        self.assertRaises(NoSuchFile,
 
51
            self.applyDeprecated,
 
52
            deprecated_in((1, 5, 0)),
 
53
            self.lockable.get, 'foo')
48
54
        self.assertRaises(NoSuchFile,
49
55
            self.applyDeprecated,
50
56
            deprecated_in((1, 5, 0)),
55
61
            self.assertEqual(4, len(unicode_string))
56
62
            byte_string = unicode_string.encode('utf-8')
57
63
            self.assertEqual(6, len(byte_string))
58
 
            self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo',
59
 
                              StringIO(unicode_string))
60
 
            self.lockable.put('foo', StringIO(byte_string))
61
 
            self.assertEqual(byte_string,
62
 
                             self.lockable.get('foo').read())
 
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())
63
78
            unicode_stream = self.applyDeprecated(
64
79
                deprecated_in((1, 5, 0)),
65
80
                self.lockable.get_utf8,
67
82
            self.assertEqual(unicode_string,
68
83
                unicode_stream.read())
69
84
            self.assertRaises(BzrBadParameterNotString,
70
 
                              self.lockable.put_utf8,
71
 
                              'bar',
72
 
                              StringIO(unicode_string)
73
 
                              )
74
 
            self.lockable.put_utf8('bar', unicode_string)
 
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)
75
95
            unicode_stream = self.applyDeprecated(
76
96
                deprecated_in((1, 5, 0)),
77
97
                self.lockable.get_utf8,
78
98
                'bar')
79
99
            self.assertEqual(unicode_string,
80
100
                unicode_stream.read())
81
 
            self.assertEqual(byte_string,
82
 
                             self.lockable.get('bar').read())
83
 
            self.lockable.put_bytes('raw', 'raw\xffbytes')
84
 
            self.assertEqual('raw\xffbytes',
85
 
                             self.lockable.get('raw').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())
86
115
        finally:
87
116
            self.lockable.unlock()
88
117
 
89
118
    def test_locks(self):
90
119
        self.lockable.lock_read()
91
120
        try:
92
 
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo', 
 
121
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
93
122
                              StringIO('bar\u1234'))
94
123
        finally:
95
124
            self.lockable.unlock()
128
157
        except NotImplementedError:
129
158
            # this lock cannot be broken
130
159
            self.lockable.unlock()
131
 
            return
 
160
            raise TestNotApplicable("%r is not breakable" % (self.lockable,))
132
161
        l2 = self.get_lockable()
133
162
        orig_factory = bzrlib.ui.ui_factory
134
163
        # silent ui - no need for stdout
151
180
            if token is not None:
152
181
                # This test does not apply, because this lockable supports
153
182
                # tokens.
154
 
                return
 
183
                raise TestNotApplicable("%r uses tokens" % (self.lockable,))
155
184
            self.assertRaises(errors.TokenLockingNotSupported,
156
185
                              self.lockable.lock_write, token='token')
157
186
        finally:
345
374
        super(TestLockableFiles_TransportLock, self).tearDown()
346
375
        # free the subtransport so that we do not get a 5 second
347
376
        # timeout due to the SFTP connection cache.
348
 
        del self.sub_transport
 
377
        try:
 
378
            del self.sub_transport
 
379
        except AttributeError:
 
380
            pass
349
381
 
350
382
    def get_lockable(self):
351
383
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)