← 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: Martin Pool
  • Date: 2009-03-24 04:49:51 UTC
  • mfrom: (3407.2.20 controlfiles)
  • mto: This revision was merged to the branch mainline in revision 4202.
  • Revision ID: mbp@sourcefrog.net-20090324044951-k5h6x6u0zz4721m5
bzrdir takes responsibility for file/directory unix modes

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_counted_lock.py
39
39
        self._lock_mode = 'r'
40
40
        self._calls.append('lock_read')
41
41
 
42
 
    def lock_write(self):
 
42
    def lock_write(self, token=None):
 
43
        if token is not None:
 
44
            if token == 'token':
 
45
                # already held by this caller
 
46
                return 'token'
 
47
            else:
 
48
                raise errors.TokenMismatch()
43
49
        self._assert_not_locked()
44
50
        self._lock_mode = 'w'
45
51
        self._calls.append('lock_write')
 
52
        return 'token'
46
53
 
47
54
    def unlock(self):
48
55
        self._assert_locked()
89
96
        real_lock.unlock()
90
97
        self.assertFalse(real_lock.is_locked())
91
98
        # lock write and unlock
92
 
        real_lock.lock_write()
 
99
        result = real_lock.lock_write()
 
100
        self.assertEqual('token', result)
93
101
        self.assertTrue(real_lock.is_locked())
94
102
        real_lock.unlock()
95
103
        self.assertFalse(real_lock.is_locked())
110
118
 
111
119
class TestCountedLock(TestCase):
112
120
 
113
 
    def test_lock_unlock(self):
 
121
    def test_read_lock(self):
114
122
        # Lock and unlock a counted lock
115
123
        real_lock = DummyLock()
116
124
        l = CountedLock(real_lock)
138
146
        l = CountedLock(real_lock)
139
147
        l.lock_write()
140
148
        l.lock_read()
141
 
        l.lock_write()
 
149
        self.assertEquals('token', l.lock_write())
142
150
        l.unlock()
143
151
        l.unlock()
144
152
        l.unlock()
159
167
            ['lock_read', 'unlock'],
160
168
            real_lock._calls)
161
169
 
 
170
    def test_write_lock_reentrant(self):
 
171
        real_lock = DummyLock()
 
172
        l = CountedLock(real_lock)
 
173
        self.assertEqual('token', l.lock_write())
 
174
        self.assertEqual('token', l.lock_write())
 
175
        l.unlock()
 
176
        l.unlock()
 
177
 
 
178
    def test_reenter_with_token(self):
 
179
        real_lock = DummyLock()
 
180
        l1 = CountedLock(real_lock)
 
181
        l2 = CountedLock(real_lock)
 
182
        token = l1.lock_write()
 
183
        self.assertEqual('token', token)
 
184
        # now imagine that we lost that connection, but we still have the
 
185
        # token...
 
186
        del l1
 
187
        # because we can supply the token, we can acquire the lock through
 
188
        # another instance
 
189
        self.assertTrue(real_lock.is_locked())
 
190
        self.assertFalse(l2.is_locked())
 
191
        self.assertEqual(token, l2.lock_write(token=token))
 
192
        self.assertTrue(l2.is_locked())
 
193
        self.assertTrue(real_lock.is_locked())
 
194
        l2.unlock()
 
195
        self.assertFalse(l2.is_locked())
 
196
        self.assertFalse(real_lock.is_locked())
 
197
 
162
198
    def test_break_lock(self):
163
199
        real_lock = DummyLock()
164
200
        l = CountedLock(real_lock)