← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_counted_lock.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_counted_lock.py
1
 
# Copyright (C) 2007, 2008, 2009, 2016 Canonical Ltd
 
1
# Copyright (C) 2007, 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for bzrlib.counted_lock"""
18
18
 
42
42
        self._calls.append('lock_read')
43
43
 
44
44
    def lock_write(self, token=None):
45
 
        if token is not None:
46
 
            if token == 'token':
47
 
                # already held by this caller
48
 
                return 'token'
49
 
            else:
50
 
                raise TokenMismatch()
 
45
        if token not in (None, 'token'):
 
46
            raise TokenMismatch(token, 'token')
51
47
        self._assert_not_locked()
52
48
        self._lock_mode = 'w'
53
49
        self._calls.append('lock_write')
107
103
        real_lock.unlock()
108
104
        self.assertFalse(real_lock.is_locked())
109
105
        # lock write and unlock
110
 
        result = real_lock.lock_write()
111
 
        self.assertEqual('token', result)
 
106
        real_lock.lock_write()
112
107
        self.assertTrue(real_lock.is_locked())
113
108
        real_lock.unlock()
114
109
        self.assertFalse(real_lock.is_locked())
129
124
 
130
125
class TestCountedLock(TestCase):
131
126
 
132
 
    def test_read_lock(self):
 
127
    def test_lock_unlock(self):
133
128
        # Lock and unlock a counted lock
134
129
        real_lock = DummyLock()
135
130
        l = CountedLock(real_lock)
143
138
        self.assertTrue(l.is_locked())
144
139
        l.unlock()
145
140
        self.assertFalse(l.is_locked())
146
 
        self.assertEqual(
 
141
        self.assertEquals(
147
142
            ['lock_read', 'unlock'],
148
143
            real_lock._calls)
149
144
 
157
152
        l = CountedLock(real_lock)
158
153
        l.lock_write()
159
154
        l.lock_read()
160
 
        self.assertEqual('token', l.lock_write())
 
155
        l.lock_write()
161
156
        l.unlock()
162
157
        l.unlock()
163
158
        l.unlock()
164
159
        self.assertFalse(l.is_locked())
165
 
        self.assertEqual(
 
160
        self.assertEquals(
166
161
            ['lock_write', 'unlock'],
167
162
            real_lock._calls)
168
163
 
174
169
        self.assertRaises(ReadOnlyError, l.lock_write)
175
170
        l.unlock()
176
171
        self.assertFalse(l.is_locked())
177
 
        self.assertEqual(
 
172
        self.assertEquals(
178
173
            ['lock_read', 'unlock'],
179
174
            real_lock._calls)
180
175
 
181
 
    def test_write_lock_reentrant(self):
182
 
        real_lock = DummyLock()
183
 
        l = CountedLock(real_lock)
184
 
        self.assertEqual('token', l.lock_write())
185
 
        self.assertEqual('token', l.lock_write())
186
 
        l.unlock()
187
 
        l.unlock()
188
 
 
189
 
    def test_reenter_with_token(self):
190
 
        real_lock = DummyLock()
191
 
        l1 = CountedLock(real_lock)
192
 
        l2 = CountedLock(real_lock)
193
 
        token = l1.lock_write()
194
 
        self.assertEqual('token', token)
195
 
        # now imagine that we lost that connection, but we still have the
196
 
        # token...
197
 
        del l1
198
 
        # because we can supply the token, we can acquire the lock through
199
 
        # another instance
200
 
        self.assertTrue(real_lock.is_locked())
201
 
        self.assertFalse(l2.is_locked())
202
 
        self.assertEqual(token, l2.lock_write(token=token))
203
 
        self.assertTrue(l2.is_locked())
204
 
        self.assertTrue(real_lock.is_locked())
205
 
        l2.unlock()
206
 
        self.assertFalse(l2.is_locked())
207
 
        self.assertFalse(real_lock.is_locked())
208
 
 
209
176
    def test_break_lock(self):
210
177
        real_lock = DummyLock()
211
178
        l = CountedLock(real_lock)
215
182
        l.break_lock()
216
183
        self.assertFalse(l.is_locked())
217
184
        self.assertFalse(real_lock.is_locked())
218
 
 
219
 
    # TODO: test get_physical_lock_status