← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: Martin Pool
  • Date: 2006-03-06 03:57:18 UTC
  • mto: This revision was merged to the branch mainline in revision 1593.
  • Revision ID: mbp@sourcefrog.net-20060306035718-a50bae3c32eec48a
Change LockDirs to format "lock-name/held/info"

For the moment, auto-create the top-level directory when it's needed.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
31
31
# concurrent actors.  This is not a typical (or necessarily supported) use;
32
32
# they're really meant for guarding between processes.
33
33
 
 
34
# These tests are run on the default transport provided by the test framework
 
35
# (typically a local disk transport).  That can be changed by the --transport
 
36
# option to bzr selftest.  The required properties of the transport
 
37
# implementation are tested separately.  (The main requirement is just that
 
38
# they don't allow overwriting nonempty directories.)
 
39
 
34
40
class TestLockDir(TestCaseWithTransport):
35
41
    """Test LockDir operations"""
36
42
 
58
64
        """Acquire and release a lock"""
59
65
        t = self.get_transport()
60
66
        lf = LockDir(t, 'test_lock')
 
67
        lf.create()
61
68
        lf.attempt_lock()
62
69
        try:
63
70
            self.assertTrue(lf.is_held)
65
72
            lf.unlock()
66
73
            self.assertFalse(lf.is_held)
67
74
 
68
 
    def test_11_lock_readonly_transport(self):
 
75
    def test_11_create_readonly_transport(self):
 
76
        """Fail to create lock on readonly transport"""
 
77
        t = self.get_readonly_transport()
 
78
        lf = LockDir(t, 'test_lock')
 
79
        self.assertRaises(UnlockableTransport, lf.create)
 
80
 
 
81
    def test_12_lock_readonly_transport(self):
69
82
        """Fail to lock on readonly transport"""
70
 
        t = self.get_readonly_transport()
71
 
        lf = LockDir(t, 'test_lock')
 
83
        lf = LockDir(self.get_transport(), 'test_lock')
 
84
        lf.create()
 
85
        lf = LockDir(self.get_readonly_transport(), 'test_lock')
72
86
        self.assertRaises(UnlockableTransport, lf.attempt_lock)
73
87
 
74
88
    def test_20_lock_contested(self):
75
89
        """Contention to get a lock"""
76
90
        t = self.get_transport()
77
91
        lf1 = LockDir(t, 'test_lock')
 
92
        lf1.create()
78
93
        lf1.attempt_lock()
79
94
        lf2 = LockDir(t, 'test_lock')
80
95
        try:
92
107
        """Peek at the state of a lock"""
93
108
        t = self.get_transport()
94
109
        lf1 = LockDir(t, 'test_lock')
 
110
        lf1.create()
95
111
        lf1.attempt_lock()
96
112
        # lock is held, should get some info on it
97
113
        info1 = lf1.peek()
108
124
        """Peek over a readonly transport"""
109
125
        t = self.get_transport()
110
126
        lf1 = LockDir(t, 'test_lock')
 
127
        lf1.create()
111
128
        lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
112
129
        self.assertEqual(lf2.peek(), None)
113
130
        lf1.attempt_lock()
124
141
        """
125
142
        t = self.get_transport()
126
143
        lf1 = LockDir(t, 'test_lock')
 
144
        lf1.create()
127
145
        lf2 = LockDir(t, 'test_lock')
128
146
        lf1.attempt_lock()
129
147
        try:
139
157
        """Succeed when waiting on a lock with no contention.
140
158
        """
141
159
        t = self.get_transport()
142
 
        lf2 = LockDir(t, 'test_lock')
 
160
        lf1 = LockDir(t, 'test_lock')
 
161
        lf1.create()
143
162
        try:
144
163
            before = time.time()
145
 
            lf2.wait_lock(timeout=0.4, poll=0.1)
 
164
            lf1.wait_lock(timeout=0.4, poll=0.1)
146
165
            after = time.time()
147
166
            self.assertTrue(after - before <= 1.0)
148
167
        finally:
149
 
            lf2.unlock()
 
168
            lf1.unlock()
150
169
 
151
170
    def test_32_lock_wait_succeed(self):
152
171
        """Succeed when trying to acquire a lock that gets released
155
174
        """
156
175
        t = self.get_transport()
157
176
        lf1 = LockDir(t, 'test_lock')
 
177
        lf1.create()
158
178
        lf1.attempt_lock()
159
179
 
160
180
        def wait_and_unlock():
181
201
        """
182
202
        t = self.get_transport()
183
203
        lf1 = LockDir(t, 'test_lock')
 
204
        lf1.create()
184
205
        lf1.attempt_lock()
185
206
 
186
207
        def wait_and_unlock():
202
223
        """Confirm a lock that's already held"""
203
224
        t = self.get_transport()
204
225
        lf1 = LockDir(t, 'test_lock')
 
226
        lf1.create()
205
227
        lf1.attempt_lock()
206
228
        lf1.confirm()
207
229
 
209
231
        """Confirm a lock that's already held"""
210
232
        t = self.get_transport()
211
233
        lf1 = LockDir(t, 'test_lock')
 
234
        lf1.create()
212
235
        self.assertRaises(LockNotHeld, lf1.confirm)
213
236
 
214
237
    def test_42_confirm_broken_manually(self):
215
238
        """Confirm a lock broken by hand"""
216
239
        t = self.get_transport()
217
240
        lf1 = LockDir(t, 'test_lock')
 
241
        lf1.create()
218
242
        lf1.attempt_lock()
219
243
        t.move('test_lock', 'lock_gone_now')
220
244
        self.assertRaises(LockBroken, lf1.confirm)
223
247
        """Break a lock whose caller has forgotten it"""
224
248
        t = self.get_transport()
225
249
        lf1 = LockDir(t, 'test_lock')
 
250
        lf1.create()
226
251
        lf1.attempt_lock()
227
252
        # we incorrectly discard the lock object without unlocking it
228
253
        del lf1
239
264
        """Lock break races with regular release"""
240
265
        t = self.get_transport()
241
266
        lf1 = LockDir(t, 'test_lock')
 
267
        lf1.create()
242
268
        lf1.attempt_lock()
243
269
        # someone else sees it's still locked
244
270
        lf2 = LockDir(t, 'test_lock')
255
281
        """Lock break races with someone else acquiring it"""
256
282
        t = self.get_transport()
257
283
        lf1 = LockDir(t, 'test_lock')
 
284
        lf1.create()
258
285
        lf1.attempt_lock()
259
286
        # someone else sees it's still locked
260
287
        lf2 = LockDir(t, 'test_lock')
271
298
    def test_46_fake_read_lock(self):
272
299
        t = self.get_transport()
273
300
        lf1 = LockDir(t, 'test_lock')
 
301
        lf1.create()
274
302
        lf1.lock_read()
275
303
        lf1.unlock()
 
304
 
 
305
    def test_50_lockdir_representation(self):
 
306
        """Check the on-disk representation of LockDirs is as expected.
 
307
 
 
308
        There should always be a top-level directory named by the lock.
 
309
        When the lock is held, there should be a lockname/held directory 
 
310
        containing an info file.
 
311
        """
 
312
        t = self.get_transport()
 
313
        lf1 = LockDir(t, 'test_lock')
 
314
        lf1.create()
 
315
        self.assertTrue(t.has('test_lock'))
 
316
        lf1.lock_write()
 
317
        self.assertTrue(t.has('test_lock/held/info'))
 
318
        lf1.unlock()
 
319
        self.assertFalse(t.has('test_lock/held/info'))