/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

Allow Repository tests to be backed onto a specific VFS as needed.

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
 
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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
from StringIO import StringIO
 
18
 
 
19
import bzrlib
 
20
from bzrlib.branch import Branch
 
21
import bzrlib.errors as errors
 
22
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
 
23
from bzrlib.lockable_files import LockableFiles, TransportLock
 
24
from bzrlib.lockdir import LockDir
 
25
from bzrlib.tests import TestCaseInTempDir
 
26
from bzrlib.tests.test_smart import TestCaseWithSmartMedium
 
27
from bzrlib.tests.test_transactions import DummyWeave
 
28
from bzrlib.transactions import (PassThroughTransaction,
 
29
                                 ReadOnlyTransaction,
 
30
                                 WriteTransaction,
 
31
                                 )
 
32
from bzrlib.transport import get_transport
 
33
 
 
34
 
 
35
# these tests are applied in each parameterized suite for LockableFiles
 
36
class _TestLockableFiles_mixin(object):
 
37
 
 
38
    def test_read_write(self):
 
39
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
 
40
        self.assertRaises(NoSuchFile, self.lockable.get_utf8, 'foo')
 
41
        self.lockable.lock_write()
 
42
        try:
 
43
            unicode_string = u'bar\u1234'
 
44
            self.assertEqual(4, len(unicode_string))
 
45
            byte_string = unicode_string.encode('utf-8')
 
46
            self.assertEqual(6, len(byte_string))
 
47
            self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo', 
 
48
                              StringIO(unicode_string))
 
49
            self.lockable.put('foo', StringIO(byte_string))
 
50
            self.assertEqual(byte_string,
 
51
                             self.lockable.get('foo').read())
 
52
            self.assertEqual(unicode_string,
 
53
                             self.lockable.get_utf8('foo').read())
 
54
            self.assertRaises(BzrBadParameterNotString,
 
55
                              self.lockable.put_utf8,
 
56
                              'bar',
 
57
                              StringIO(unicode_string)
 
58
                              )
 
59
            self.lockable.put_utf8('bar', unicode_string)
 
60
            self.assertEqual(unicode_string, 
 
61
                             self.lockable.get_utf8('bar').read())
 
62
            self.assertEqual(byte_string, 
 
63
                             self.lockable.get('bar').read())
 
64
        finally:
 
65
            self.lockable.unlock()
 
66
 
 
67
    def test_locks(self):
 
68
        self.lockable.lock_read()
 
69
        try:
 
70
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo', 
 
71
                              StringIO('bar\u1234'))
 
72
        finally:
 
73
            self.lockable.unlock()
 
74
 
 
75
    def test_transactions(self):
 
76
        self.assertIs(self.lockable.get_transaction().__class__,
 
77
                      PassThroughTransaction)
 
78
        self.lockable.lock_read()
 
79
        try:
 
80
            self.assertIs(self.lockable.get_transaction().__class__,
 
81
                          ReadOnlyTransaction)
 
82
        finally:
 
83
            self.lockable.unlock()
 
84
        self.assertIs(self.lockable.get_transaction().__class__,
 
85
                      PassThroughTransaction)
 
86
        self.lockable.lock_write()
 
87
        self.assertIs(self.lockable.get_transaction().__class__,
 
88
                      WriteTransaction)
 
89
        # check that finish is called:
 
90
        vf = DummyWeave('a')
 
91
        self.lockable.get_transaction().register_dirty(vf)
 
92
        self.lockable.unlock()
 
93
        self.assertTrue(vf.finished)
 
94
 
 
95
    def test__escape(self):
 
96
        self.assertEqual('%25', self.lockable._escape('%'))
 
97
        
 
98
    def test__escape_empty(self):
 
99
        self.assertEqual('', self.lockable._escape(''))
 
100
 
 
101
    def test_break_lock(self):
 
102
        # some locks are not breakable
 
103
        self.lockable.lock_write()
 
104
        try:
 
105
            self.assertRaises(AssertionError, self.lockable.break_lock)
 
106
        except NotImplementedError:
 
107
            # this lock cannot be broken
 
108
            self.lockable.unlock()
 
109
            return
 
110
        l2 = self.get_lockable()
 
111
        orig_factory = bzrlib.ui.ui_factory
 
112
        # silent ui - no need for stdout
 
113
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
114
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
 
115
        try:
 
116
            l2.break_lock()
 
117
        finally:
 
118
            bzrlib.ui.ui_factory = orig_factory
 
119
        try:
 
120
            l2.lock_write()
 
121
            l2.unlock()
 
122
        finally:
 
123
            self.assertRaises(errors.LockBroken, self.lockable.unlock)
 
124
            self.assertFalse(self.lockable.is_locked())
 
125
 
 
126
 
 
127
# This method of adapting tests to parameters is different to 
 
128
# the TestProviderAdapters used elsewhere, but seems simpler for this 
 
129
# case.  
 
130
class TestLockableFiles_TransportLock(TestCaseInTempDir,
 
131
                                      _TestLockableFiles_mixin):
 
132
 
 
133
    def setUp(self):
 
134
        super(TestLockableFiles_TransportLock, self).setUp()
 
135
        transport = get_transport('.')
 
136
        transport.mkdir('.bzr')
 
137
        self.sub_transport = transport.clone('.bzr')
 
138
        self.lockable = self.get_lockable()
 
139
        self.lockable.create_lock()
 
140
 
 
141
    def tearDown(self):
 
142
        super(TestLockableFiles_TransportLock, self).tearDown()
 
143
        # free the subtransport so that we do not get a 5 second
 
144
        # timeout due to the SFTP connection cache.
 
145
        del self.sub_transport
 
146
 
 
147
    def get_lockable(self):
 
148
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)
 
149
        
 
150
 
 
151
class TestLockableFiles_LockDir(TestCaseInTempDir,
 
152
                              _TestLockableFiles_mixin):
 
153
    """LockableFile tests run with LockDir underneath"""
 
154
 
 
155
    def setUp(self):
 
156
        super(TestLockableFiles_LockDir, self).setUp()
 
157
        self.transport = get_transport('.')
 
158
        self.lockable = self.get_lockable()
 
159
        # the lock creation here sets mode - test_permissions on branch 
 
160
        # tests that implicitly, but it might be a good idea to factor 
 
161
        # out the mode checking logic and have it applied to loackable files
 
162
        # directly. RBC 20060418
 
163
        self.lockable.create_lock()
 
164
 
 
165
    def get_lockable(self):
 
166
        return LockableFiles(self.transport, 'my-lock', LockDir)
 
167
 
 
168
    def test_lock_created(self):
 
169
        self.assertTrue(self.transport.has('my-lock'))
 
170
        self.lockable.lock_write()
 
171
        self.assertTrue(self.transport.has('my-lock/held/info'))
 
172
        self.lockable.unlock()
 
173
        self.assertFalse(self.transport.has('my-lock/held/info'))
 
174
        self.assertTrue(self.transport.has('my-lock'))
 
175
 
 
176
 
 
177
    # TODO: Test the lockdir inherits the right file and directory permissions
 
178
    # from the LockableFiles.
 
179
        
 
180
 
 
181
class TestLockableFiles_RemoteLockDir(TestCaseWithSmartMedium,
 
182
                              _TestLockableFiles_mixin):
 
183
    """LockableFile tests run with RemoteLockDir on a branch."""
 
184
 
 
185
    def setUp(self):
 
186
        super(TestLockableFiles_RemoteLockDir, self).setUp()
 
187
        # can only get a RemoteLockDir with some RemoteObject...
 
188
        # use a branch as thats what we want. These mixin tests test the end
 
189
        # to end behaviour, so stubbing out the backend and simulating would
 
190
        # defeat the purpose. We test the protocol implementation separately
 
191
        # in test_remote and test_smart as usual.
 
192
        self.make_branch('foo')
 
193
        self.transport = get_transport('.')
 
194
        self.lockable = self.get_lockable()
 
195
 
 
196
    def get_lockable(self):
 
197
        # getting a new lockable involves opening a new instance of the branch
 
198
        branch = bzrlib.branch.Branch.open(self.get_url('foo'))
 
199
        return branch.control_files