/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_lockable_files.py

  • Committer: Andrew Bennetts
  • Date: 2008-09-08 12:59:00 UTC
  • mfrom: (3695 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3756.
  • Revision ID: andrew.bennetts@canonical.com-20080908125900-8ywtsr7jqyyatjz0
Merge from bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2008 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
24
24
    )
25
25
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
26
26
from bzrlib.lockable_files import LockableFiles, TransportLock
27
 
from bzrlib.tests import TestCaseInTempDir
 
27
from bzrlib.symbol_versioning import (
 
28
    deprecated_in,
 
29
    )
 
30
from bzrlib.tests import (
 
31
    TestCaseInTempDir,
 
32
    TestNotApplicable,
 
33
    )
28
34
from bzrlib.tests.test_smart import TestCaseWithSmartMedium
29
35
from bzrlib.tests.test_transactions import DummyWeave
30
36
from bzrlib.transactions import (PassThroughTransaction,
35
41
 
36
42
 
37
43
# these tests are applied in each parameterized suite for LockableFiles
 
44
#
 
45
# they use an old style of parameterization, but we want to remove this class
 
46
# so won't modernize them now. - mbp 20080430
38
47
class _TestLockableFiles_mixin(object):
39
48
 
40
49
    def test_read_write(self):
41
 
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
42
 
        self.assertRaises(NoSuchFile, self.lockable.get_utf8, 'foo')
 
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')
43
58
        self.lockable.lock_write()
44
59
        try:
45
60
            unicode_string = u'bar\u1234'
46
61
            self.assertEqual(4, len(unicode_string))
47
62
            byte_string = unicode_string.encode('utf-8')
48
63
            self.assertEqual(6, len(byte_string))
49
 
            self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo',
50
 
                              StringIO(unicode_string))
51
 
            self.lockable.put('foo', StringIO(byte_string))
52
 
            self.assertEqual(byte_string,
53
 
                             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())
 
78
            unicode_stream = self.applyDeprecated(
 
79
                deprecated_in((1, 5, 0)),
 
80
                self.lockable.get_utf8,
 
81
                'foo')
54
82
            self.assertEqual(unicode_string,
55
 
                             self.lockable.get_utf8('foo').read())
 
83
                unicode_stream.read())
56
84
            self.assertRaises(BzrBadParameterNotString,
57
 
                              self.lockable.put_utf8,
58
 
                              'bar',
59
 
                              StringIO(unicode_string)
60
 
                              )
61
 
            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)
 
95
            unicode_stream = self.applyDeprecated(
 
96
                deprecated_in((1, 5, 0)),
 
97
                self.lockable.get_utf8,
 
98
                'bar')
62
99
            self.assertEqual(unicode_string,
63
 
                             self.lockable.get_utf8('bar').read())
64
 
            self.assertEqual(byte_string,
65
 
                             self.lockable.get('bar').read())
66
 
            self.lockable.put_bytes('raw', 'raw\xffbytes')
67
 
            self.assertEqual('raw\xffbytes',
68
 
                             self.lockable.get('raw').read())
 
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())
69
115
        finally:
70
116
            self.lockable.unlock()
71
117
 
72
118
    def test_locks(self):
73
119
        self.lockable.lock_read()
74
120
        try:
75
 
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo', 
 
121
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo',
76
122
                              StringIO('bar\u1234'))
77
123
        finally:
78
124
            self.lockable.unlock()
111
157
        except NotImplementedError:
112
158
            # this lock cannot be broken
113
159
            self.lockable.unlock()
114
 
            return
 
160
            raise TestNotApplicable("%r is not breakable" % (self.lockable,))
115
161
        l2 = self.get_lockable()
116
162
        orig_factory = bzrlib.ui.ui_factory
117
163
        # silent ui - no need for stdout
134
180
            if token is not None:
135
181
                # This test does not apply, because this lockable supports
136
182
                # tokens.
137
 
                return
 
183
                raise TestNotApplicable("%r uses tokens" % (self.lockable,))
138
184
            self.assertRaises(errors.TokenLockingNotSupported,
139
185
                              self.lockable.lock_write, token='token')
140
186
        finally:
328
374
        super(TestLockableFiles_TransportLock, self).tearDown()
329
375
        # free the subtransport so that we do not get a 5 second
330
376
        # timeout due to the SFTP connection cache.
331
 
        del self.sub_transport
 
377
        try:
 
378
            del self.sub_transport
 
379
        except AttributeError:
 
380
            pass
332
381
 
333
382
    def get_lockable(self):
334
383
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)