← 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: Canonical.com Patch Queue Manager
  • Date: 2009-03-24 13:40:38 UTC
  • mfrom: (3193.8.32 guess-renames)
  • Revision ID: pqm@pqm.ubuntu.com-20090324134038-bfrsas8eleeoiv0a
(abentley) Add guess-renames command.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_counted_lock.py
1
 
# Copyright (C) 2007, 2008, 2009 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
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)
157
152
        l = CountedLock(real_lock)
158
153
        l.lock_write()
159
154
        l.lock_read()
160
 
        self.assertEquals('token', l.lock_write())
 
155
        l.lock_write()
161
156
        l.unlock()
162
157
        l.unlock()
163
158
        l.unlock()
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)