← 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: John Arbash Meinel
  • Date: 2011-05-11 11:35:28 UTC
  • mto: This revision was merged to the branch mainline in revision 5851.
  • Revision ID: john@arbash-meinel.com-20110511113528-qepibuwxicjrbb2h
Break compatibility with python <2.6.

This includes auditing the code for places where we were doing
explicit 'sys.version' checks and removing them as appropriate.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
 
1
# Copyright (C) 2006-2011 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Tests for LockDir"""
 
18
 
 
19
import os
 
20
import time
 
21
 
 
22
import bzrlib
 
23
from bzrlib import (
 
24
    config,
 
25
    errors,
 
26
    lock,
 
27
    osutils,
 
28
    tests,
 
29
    transport,
 
30
    )
 
31
from bzrlib.errors import (
 
32
    LockBreakMismatch,
 
33
    LockBroken,
 
34
    LockContention,
 
35
    LockFailed,
 
36
    LockNotHeld,
 
37
    )
 
38
from bzrlib.lockdir import LockDir
 
39
from bzrlib.tests import (
 
40
    features,
 
41
    TestCaseWithTransport,
 
42
    )
 
43
from bzrlib.trace import note
 
44
 
 
45
# These tests are run on the default transport provided by the test framework
 
46
# (typically a local disk transport).  That can be changed by the --transport
 
47
# option to bzr selftest.  The required properties of the transport
 
48
# implementation are tested separately.  (The main requirement is just that
 
49
# they don't allow overwriting nonempty directories.)
 
50
 
 
51
class TestLockDir(TestCaseWithTransport):
 
52
    """Test LockDir operations"""
 
53
 
 
54
    def logging_report_function(self, fmt, *args):
 
55
        self._logged_reports.append((fmt, args))
 
56
 
 
57
    def setup_log_reporter(self, lock_dir):
 
58
        self._logged_reports = []
 
59
        lock_dir._report_function = self.logging_report_function
 
60
 
 
61
    def test_00_lock_creation(self):
 
62
        """Creation of lock file on a transport"""
 
63
        t = self.get_transport()
 
64
        lf = LockDir(t, 'test_lock')
 
65
        self.assertFalse(lf.is_held)
 
66
 
 
67
    def test_01_lock_repr(self):
 
68
        """Lock string representation"""
 
69
        lf = LockDir(self.get_transport(), 'test_lock')
 
70
        r = repr(lf)
 
71
        self.assertContainsRe(r, r'^LockDir\(.*/test_lock\)$')
 
72
 
 
73
    def test_02_unlocked_peek(self):
 
74
        lf = LockDir(self.get_transport(), 'test_lock')
 
75
        self.assertEqual(lf.peek(), None)
 
76
 
 
77
    def get_lock(self):
 
78
        return LockDir(self.get_transport(), 'test_lock')
 
79
 
 
80
    def test_unlock_after_break_raises(self):
 
81
        ld = self.get_lock()
 
82
        ld2 = self.get_lock()
 
83
        ld.create()
 
84
        ld.attempt_lock()
 
85
        ld2.force_break(ld2.peek())
 
86
        self.assertRaises(LockBroken, ld.unlock)
 
87
 
 
88
    def test_03_readonly_peek(self):
 
89
        lf = LockDir(self.get_readonly_transport(), 'test_lock')
 
90
        self.assertEqual(lf.peek(), None)
 
91
 
 
92
    def test_10_lock_uncontested(self):
 
93
        """Acquire and release a lock"""
 
94
        t = self.get_transport()
 
95
        lf = LockDir(t, 'test_lock')
 
96
        lf.create()
 
97
        lf.attempt_lock()
 
98
        try:
 
99
            self.assertTrue(lf.is_held)
 
100
        finally:
 
101
            lf.unlock()
 
102
            self.assertFalse(lf.is_held)
 
103
 
 
104
    def test_11_create_readonly_transport(self):
 
105
        """Fail to create lock on readonly transport"""
 
106
        t = self.get_readonly_transport()
 
107
        lf = LockDir(t, 'test_lock')
 
108
        self.assertRaises(LockFailed, lf.create)
 
109
 
 
110
    def test_12_lock_readonly_transport(self):
 
111
        """Fail to lock on readonly transport"""
 
112
        lf = LockDir(self.get_transport(), 'test_lock')
 
113
        lf.create()
 
114
        lf = LockDir(self.get_readonly_transport(), 'test_lock')
 
115
        self.assertRaises(LockFailed, lf.attempt_lock)
 
116
 
 
117
    def test_20_lock_contested(self):
 
118
        """Contention to get a lock"""
 
119
        t = self.get_transport()
 
120
        lf1 = LockDir(t, 'test_lock')
 
121
        lf1.create()
 
122
        lf1.attempt_lock()
 
123
        lf2 = LockDir(t, 'test_lock')
 
124
        try:
 
125
            # locking is between LockDir instances; aliases within
 
126
            # a single process are not detected
 
127
            lf2.attempt_lock()
 
128
            self.fail('Failed to detect lock collision')
 
129
        except LockContention, e:
 
130
            self.assertEqual(e.lock, lf2)
 
131
            self.assertContainsRe(str(e),
 
132
                    r'^Could not acquire.*test_lock.*$')
 
133
        lf1.unlock()
 
134
 
 
135
    def test_20_lock_peek(self):
 
136
        """Peek at the state of a lock"""
 
137
        t = self.get_transport()
 
138
        lf1 = LockDir(t, 'test_lock')
 
139
        lf1.create()
 
140
        lf1.attempt_lock()
 
141
        self.addCleanup(lf1.unlock)
 
142
        # lock is held, should get some info on it
 
143
        info1 = lf1.peek()
 
144
        self.assertEqual(set(info1.keys()),
 
145
                         set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
 
146
        # should get the same info if we look at it through a different
 
147
        # instance
 
148
        info2 = LockDir(t, 'test_lock').peek()
 
149
        self.assertEqual(info1, info2)
 
150
        # locks which are never used should be not-held
 
151
        self.assertEqual(LockDir(t, 'other_lock').peek(), None)
 
152
 
 
153
    def test_21_peek_readonly(self):
 
154
        """Peek over a readonly transport"""
 
155
        t = self.get_transport()
 
156
        lf1 = LockDir(t, 'test_lock')
 
157
        lf1.create()
 
158
        lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
 
159
        self.assertEqual(lf2.peek(), None)
 
160
        lf1.attempt_lock()
 
161
        self.addCleanup(lf1.unlock)
 
162
        info2 = lf2.peek()
 
163
        self.assertTrue(info2)
 
164
        self.assertEqual(info2['nonce'], lf1.nonce)
 
165
 
 
166
    def test_30_lock_wait_fail(self):
 
167
        """Wait on a lock, then fail
 
168
 
 
169
        We ask to wait up to 400ms; this should fail within at most one
 
170
        second.  (Longer times are more realistic but we don't want the test
 
171
        suite to take too long, and this should do for now.)
 
172
        """
 
173
        t = self.get_transport()
 
174
        lf1 = LockDir(t, 'test_lock')
 
175
        lf1.create()
 
176
        lf2 = LockDir(t, 'test_lock')
 
177
        self.setup_log_reporter(lf2)
 
178
        lf1.attempt_lock()
 
179
        try:
 
180
            before = time.time()
 
181
            self.assertRaises(LockContention, lf2.wait_lock,
 
182
                              timeout=0.4, poll=0.1)
 
183
            after = time.time()
 
184
            # it should only take about 0.4 seconds, but we allow more time in
 
185
            # case the machine is heavily loaded
 
186
            self.assertTrue(after - before <= 8.0,
 
187
                    "took %f seconds to detect lock contention" % (after - before))
 
188
        finally:
 
189
            lf1.unlock()
 
190
        self.assertEqual(1, len(self._logged_reports))
 
191
        self.assertEqual(self._logged_reports[0][0],
 
192
            '%s lock %s held by %s\n'
 
193
            'at %s [process #%s], acquired %s.\n'
 
194
            'Will continue to try until %s, unless '
 
195
            'you press Ctrl-C.\n'
 
196
            'See "bzr help break-lock" for more.')
 
197
        start, lock_url, user, hostname, pid, time_ago, deadline_str = \
 
198
            self._logged_reports[0][1]
 
199
        self.assertEqual(start, u'Unable to obtain')
 
200
        self.assertEqual(user, u'jrandom@example.com')
 
201
        # skip hostname
 
202
        self.assertContainsRe(pid, r'\d+')
 
203
        self.assertContainsRe(time_ago, r'.* ago')
 
204
        self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
 
205
 
 
206
    def test_31_lock_wait_easy(self):
 
207
        """Succeed when waiting on a lock with no contention.
 
208
        """
 
209
        t = self.get_transport()
 
210
        lf1 = LockDir(t, 'test_lock')
 
211
        lf1.create()
 
212
        self.setup_log_reporter(lf1)
 
213
        try:
 
214
            before = time.time()
 
215
            lf1.wait_lock(timeout=0.4, poll=0.1)
 
216
            after = time.time()
 
217
            self.assertTrue(after - before <= 1.0)
 
218
        finally:
 
219
            lf1.unlock()
 
220
        self.assertEqual([], self._logged_reports)
 
221
 
 
222
    def test_40_confirm_easy(self):
 
223
        """Confirm a lock that's already held"""
 
224
        t = self.get_transport()
 
225
        lf1 = LockDir(t, 'test_lock')
 
226
        lf1.create()
 
227
        lf1.attempt_lock()
 
228
        self.addCleanup(lf1.unlock)
 
229
        lf1.confirm()
 
230
 
 
231
    def test_41_confirm_not_held(self):
 
232
        """Confirm a lock that's already held"""
 
233
        t = self.get_transport()
 
234
        lf1 = LockDir(t, 'test_lock')
 
235
        lf1.create()
 
236
        self.assertRaises(LockNotHeld, lf1.confirm)
 
237
 
 
238
    def test_42_confirm_broken_manually(self):
 
239
        """Confirm a lock broken by hand"""
 
240
        t = self.get_transport()
 
241
        lf1 = LockDir(t, 'test_lock')
 
242
        lf1.create()
 
243
        lf1.attempt_lock()
 
244
        t.move('test_lock', 'lock_gone_now')
 
245
        self.assertRaises(LockBroken, lf1.confirm)
 
246
        # Clean up
 
247
        t.move('lock_gone_now', 'test_lock')
 
248
        lf1.unlock()
 
249
 
 
250
    def test_43_break(self):
 
251
        """Break a lock whose caller has forgotten it"""
 
252
        t = self.get_transport()
 
253
        lf1 = LockDir(t, 'test_lock')
 
254
        lf1.create()
 
255
        lf1.attempt_lock()
 
256
        # we incorrectly discard the lock object without unlocking it
 
257
        del lf1
 
258
        # someone else sees it's still locked
 
259
        lf2 = LockDir(t, 'test_lock')
 
260
        holder_info = lf2.peek()
 
261
        self.assertTrue(holder_info)
 
262
        lf2.force_break(holder_info)
 
263
        # now we should be able to take it
 
264
        lf2.attempt_lock()
 
265
        self.addCleanup(lf2.unlock)
 
266
        lf2.confirm()
 
267
 
 
268
    def test_44_break_already_released(self):
 
269
        """Lock break races with regular release"""
 
270
        t = self.get_transport()
 
271
        lf1 = LockDir(t, 'test_lock')
 
272
        lf1.create()
 
273
        lf1.attempt_lock()
 
274
        # someone else sees it's still locked
 
275
        lf2 = LockDir(t, 'test_lock')
 
276
        holder_info = lf2.peek()
 
277
        # in the interim the lock is released
 
278
        lf1.unlock()
 
279
        # break should succeed
 
280
        lf2.force_break(holder_info)
 
281
        # now we should be able to take it
 
282
        lf2.attempt_lock()
 
283
        self.addCleanup(lf2.unlock)
 
284
        lf2.confirm()
 
285
 
 
286
    def test_45_break_mismatch(self):
 
287
        """Lock break races with someone else acquiring it"""
 
288
        t = self.get_transport()
 
289
        lf1 = LockDir(t, 'test_lock')
 
290
        lf1.create()
 
291
        lf1.attempt_lock()
 
292
        # someone else sees it's still locked
 
293
        lf2 = LockDir(t, 'test_lock')
 
294
        holder_info = lf2.peek()
 
295
        # in the interim the lock is released
 
296
        lf1.unlock()
 
297
        lf3 = LockDir(t, 'test_lock')
 
298
        lf3.attempt_lock()
 
299
        # break should now *fail*
 
300
        self.assertRaises(LockBreakMismatch, lf2.force_break,
 
301
                          holder_info)
 
302
        lf3.unlock()
 
303
 
 
304
    def test_46_fake_read_lock(self):
 
305
        t = self.get_transport()
 
306
        lf1 = LockDir(t, 'test_lock')
 
307
        lf1.create()
 
308
        lf1.lock_read()
 
309
        lf1.unlock()
 
310
 
 
311
    def test_50_lockdir_representation(self):
 
312
        """Check the on-disk representation of LockDirs is as expected.
 
313
 
 
314
        There should always be a top-level directory named by the lock.
 
315
        When the lock is held, there should be a lockname/held directory
 
316
        containing an info file.
 
317
        """
 
318
        t = self.get_transport()
 
319
        lf1 = LockDir(t, 'test_lock')
 
320
        lf1.create()
 
321
        self.assertTrue(t.has('test_lock'))
 
322
        lf1.lock_write()
 
323
        self.assertTrue(t.has('test_lock/held/info'))
 
324
        lf1.unlock()
 
325
        self.assertFalse(t.has('test_lock/held/info'))
 
326
 
 
327
    def test_break_lock(self):
 
328
        # the ui based break_lock routine should Just Work (tm)
 
329
        ld1 = self.get_lock()
 
330
        ld2 = self.get_lock()
 
331
        ld1.create()
 
332
        ld1.lock_write()
 
333
        # do this without IO redirection to ensure it doesn't prompt.
 
334
        self.assertRaises(AssertionError, ld1.break_lock)
 
335
        orig_factory = bzrlib.ui.ui_factory
 
336
        bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
 
337
        try:
 
338
            ld2.break_lock()
 
339
            self.assertRaises(LockBroken, ld1.unlock)
 
340
        finally:
 
341
            bzrlib.ui.ui_factory = orig_factory
 
342
 
 
343
    def test_break_lock_corrupt_info(self):
 
344
        """break_lock works even if the info file is corrupt (and tells the UI
 
345
        that it is corrupt).
 
346
        """
 
347
        ld = self.get_lock()
 
348
        ld2 = self.get_lock()
 
349
        ld.create()
 
350
        ld.lock_write()
 
351
        ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
 
352
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
 
353
            def __init__(self):
 
354
                self.prompts = []
 
355
            def get_boolean(self, prompt):
 
356
                self.prompts.append(('boolean', prompt))
 
357
                return True
 
358
        ui = LoggingUIFactory()
 
359
        self.overrideAttr(bzrlib.ui, 'ui_factory', ui)
 
360
        ld2.break_lock()
 
361
        self.assertLength(1, ui.prompts)
 
362
        self.assertEqual('boolean', ui.prompts[0][0])
 
363
        self.assertStartsWith(ui.prompts[0][1], 'Break (corrupt LockDir')
 
364
        self.assertRaises(LockBroken, ld.unlock)
 
365
 
 
366
    def test_break_lock_missing_info(self):
 
367
        """break_lock works even if the info file is missing (and tells the UI
 
368
        that it is corrupt).
 
369
        """
 
370
        ld = self.get_lock()
 
371
        ld2 = self.get_lock()
 
372
        ld.create()
 
373
        ld.lock_write()
 
374
        ld.transport.delete('test_lock/held/info')
 
375
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
 
376
            def __init__(self):
 
377
                self.prompts = []
 
378
            def get_boolean(self, prompt):
 
379
                self.prompts.append(('boolean', prompt))
 
380
                return True
 
381
        ui = LoggingUIFactory()
 
382
        orig_factory = bzrlib.ui.ui_factory
 
383
        bzrlib.ui.ui_factory = ui
 
384
        try:
 
385
            ld2.break_lock()
 
386
            self.assertRaises(LockBroken, ld.unlock)
 
387
            self.assertLength(0, ui.prompts)
 
388
        finally:
 
389
            bzrlib.ui.ui_factory = orig_factory
 
390
        # Suppress warnings due to ld not being unlocked
 
391
        # XXX: if lock_broken hook was invoked in this case, this hack would
 
392
        # not be necessary.  - Andrew Bennetts, 2010-09-06.
 
393
        del self._lock_actions[:]
 
394
 
 
395
    def test_create_missing_base_directory(self):
 
396
        """If LockDir.path doesn't exist, it can be created
 
397
 
 
398
        Some people manually remove the entire lock/ directory trying
 
399
        to unlock a stuck repository/branch/etc. Rather than failing
 
400
        after that, just create the lock directory when needed.
 
401
        """
 
402
        t = self.get_transport()
 
403
        lf1 = LockDir(t, 'test_lock')
 
404
 
 
405
        lf1.create()
 
406
        self.assertTrue(t.has('test_lock'))
 
407
 
 
408
        t.rmdir('test_lock')
 
409
        self.assertFalse(t.has('test_lock'))
 
410
 
 
411
        # This will create 'test_lock' if it needs to
 
412
        lf1.lock_write()
 
413
        self.assertTrue(t.has('test_lock'))
 
414
        self.assertTrue(t.has('test_lock/held/info'))
 
415
 
 
416
        lf1.unlock()
 
417
        self.assertFalse(t.has('test_lock/held/info'))
 
418
 
 
419
    def test__format_lock_info(self):
 
420
        ld1 = self.get_lock()
 
421
        ld1.create()
 
422
        ld1.lock_write()
 
423
        try:
 
424
            info_list = ld1._format_lock_info(ld1.peek())
 
425
        finally:
 
426
            ld1.unlock()
 
427
        self.assertEqual(info_list[0], u'jrandom@example.com')
 
428
        # info_list[1] is hostname. we skip this.
 
429
        self.assertContainsRe(info_list[2], '^\d+$') # pid
 
430
        self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
 
431
 
 
432
    def test_lock_without_email(self):
 
433
        global_config = config.GlobalConfig()
 
434
        # Intentionally has no email address
 
435
        global_config.set_user_option('email', 'User Identity')
 
436
        ld1 = self.get_lock()
 
437
        ld1.create()
 
438
        ld1.lock_write()
 
439
        ld1.unlock()
 
440
 
 
441
    def test_lock_permission(self):
 
442
        self.requireFeature(features.not_running_as_root)
 
443
        if not osutils.supports_posix_readonly():
 
444
            raise tests.TestSkipped('Cannot induce a permission failure')
 
445
        ld1 = self.get_lock()
 
446
        lock_path = ld1.transport.local_abspath('test_lock')
 
447
        os.mkdir(lock_path)
 
448
        osutils.make_readonly(lock_path)
 
449
        self.assertRaises(errors.LockFailed, ld1.attempt_lock)
 
450
 
 
451
    def test_lock_by_token(self):
 
452
        ld1 = self.get_lock()
 
453
        token = ld1.lock_write()
 
454
        self.addCleanup(ld1.unlock)
 
455
        self.assertNotEqual(None, token)
 
456
        ld2 = self.get_lock()
 
457
        t2 = ld2.lock_write(token)
 
458
        self.addCleanup(ld2.unlock)
 
459
        self.assertEqual(token, t2)
 
460
 
 
461
    def test_lock_with_buggy_rename(self):
 
462
        # test that lock acquisition handles servers which pretend they
 
463
        # renamed correctly but that actually fail
 
464
        t = transport.get_transport('brokenrename+' + self.get_url())
 
465
        ld1 = LockDir(t, 'test_lock')
 
466
        ld1.create()
 
467
        ld1.attempt_lock()
 
468
        ld2 = LockDir(t, 'test_lock')
 
469
        # we should fail to lock
 
470
        e = self.assertRaises(errors.LockContention, ld2.attempt_lock)
 
471
        # now the original caller should succeed in unlocking
 
472
        ld1.unlock()
 
473
        # and there should be nothing left over
 
474
        self.assertEquals([], t.list_dir('test_lock'))
 
475
 
 
476
    def test_failed_lock_leaves_no_trash(self):
 
477
        # if we fail to acquire the lock, we don't leave pending directories
 
478
        # behind -- https://bugs.launchpad.net/bzr/+bug/109169
 
479
        ld1 = self.get_lock()
 
480
        ld2 = self.get_lock()
 
481
        # should be nothing before we start
 
482
        ld1.create()
 
483
        t = self.get_transport().clone('test_lock')
 
484
        def check_dir(a):
 
485
            self.assertEquals(a, t.list_dir('.'))
 
486
        check_dir([])
 
487
        # when held, that's all we see
 
488
        ld1.attempt_lock()
 
489
        self.addCleanup(ld1.unlock)
 
490
        check_dir(['held'])
 
491
        # second guy should fail
 
492
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
 
493
        # no kibble
 
494
        check_dir(['held'])
 
495
 
 
496
    def test_no_lockdir_info(self):
 
497
        """We can cope with empty info files."""
 
498
        # This seems like a fairly common failure case - see
 
499
        # <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
 
500
        # Processes are often interrupted after opening the file
 
501
        # before the actual contents are committed.
 
502
        t = self.get_transport()
 
503
        t.mkdir('test_lock')
 
504
        t.mkdir('test_lock/held')
 
505
        t.put_bytes('test_lock/held/info', '')
 
506
        lf = LockDir(t, 'test_lock')
 
507
        info = lf.peek()
 
508
        formatted_info = lf._format_lock_info(info)
 
509
        self.assertEquals(
 
510
            ['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
 
511
            formatted_info)
 
512
 
 
513
    def test_corrupt_lockdir_info(self):
 
514
        """We can cope with corrupt (and thus unparseable) info files."""
 
515
        # This seems like a fairly common failure case too - see
 
516
        # <https://bugs.launchpad.net/bzr/+bug/619872> for instance.
 
517
        # In particular some systems tend to fill recently created files with
 
518
        # nul bytes after recovering from a system crash.
 
519
        t = self.get_transport()
 
520
        t.mkdir('test_lock')
 
521
        t.mkdir('test_lock/held')
 
522
        t.put_bytes('test_lock/held/info', '\0')
 
523
        lf = LockDir(t, 'test_lock')
 
524
        self.assertRaises(errors.LockCorrupt, lf.peek)
 
525
        # Currently attempt_lock gives LockContention, but LockCorrupt would be
 
526
        # a reasonable result too.
 
527
        self.assertRaises(
 
528
            (errors.LockCorrupt, errors.LockContention), lf.attempt_lock)
 
529
        self.assertRaises(errors.LockCorrupt, lf.validate_token, 'fake token')
 
530
 
 
531
    def test_missing_lockdir_info(self):
 
532
        """We can cope with absent info files."""
 
533
        t = self.get_transport()
 
534
        t.mkdir('test_lock')
 
535
        t.mkdir('test_lock/held')
 
536
        lf = LockDir(t, 'test_lock')
 
537
        # In this case we expect the 'not held' result from peek, because peek
 
538
        # cannot be expected to notice that there is a 'held' directory with no
 
539
        # 'info' file.
 
540
        self.assertEqual(None, lf.peek())
 
541
        # And lock/unlock may work or give LockContention (but not any other
 
542
        # error).
 
543
        try:
 
544
            lf.attempt_lock()
 
545
        except LockContention:
 
546
            # LockContention is ok, and expected on Windows
 
547
            pass
 
548
        else:
 
549
            # no error is ok, and expected on POSIX (because POSIX allows
 
550
            # os.rename over an empty directory).
 
551
            lf.unlock()
 
552
        # Currently raises TokenMismatch, but LockCorrupt would be reasonable
 
553
        # too.
 
554
        self.assertRaises(
 
555
            (errors.TokenMismatch, errors.LockCorrupt),
 
556
            lf.validate_token, 'fake token')
 
557
 
 
558
 
 
559
class TestLockDirHooks(TestCaseWithTransport):
 
560
 
 
561
    def setUp(self):
 
562
        super(TestLockDirHooks, self).setUp()
 
563
        self._calls = []
 
564
 
 
565
    def get_lock(self):
 
566
        return LockDir(self.get_transport(), 'test_lock')
 
567
 
 
568
    def record_hook(self, result):
 
569
        self._calls.append(result)
 
570
 
 
571
    def test_LockDir_acquired_success(self):
 
572
        # the LockDir.lock_acquired hook fires when a lock is acquired.
 
573
        LockDir.hooks.install_named_hook('lock_acquired',
 
574
                                         self.record_hook, 'record_hook')
 
575
        ld = self.get_lock()
 
576
        ld.create()
 
577
        self.assertEqual([], self._calls)
 
578
        result = ld.attempt_lock()
 
579
        lock_path = ld.transport.abspath(ld.path)
 
580
        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
 
581
        ld.unlock()
 
582
        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
 
583
 
 
584
    def test_LockDir_acquired_fail(self):
 
585
        # the LockDir.lock_acquired hook does not fire on failure.
 
586
        ld = self.get_lock()
 
587
        ld.create()
 
588
        ld2 = self.get_lock()
 
589
        ld2.attempt_lock()
 
590
        # install a lock hook now, when the disk lock is locked
 
591
        LockDir.hooks.install_named_hook('lock_acquired',
 
592
                                         self.record_hook, 'record_hook')
 
593
        self.assertRaises(errors.LockContention, ld.attempt_lock)
 
594
        self.assertEqual([], self._calls)
 
595
        ld2.unlock()
 
596
        self.assertEqual([], self._calls)
 
597
 
 
598
    def test_LockDir_released_success(self):
 
599
        # the LockDir.lock_released hook fires when a lock is acquired.
 
600
        LockDir.hooks.install_named_hook('lock_released',
 
601
                                         self.record_hook, 'record_hook')
 
602
        ld = self.get_lock()
 
603
        ld.create()
 
604
        self.assertEqual([], self._calls)
 
605
        result = ld.attempt_lock()
 
606
        self.assertEqual([], self._calls)
 
607
        ld.unlock()
 
608
        lock_path = ld.transport.abspath(ld.path)
 
609
        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
 
610
 
 
611
    def test_LockDir_released_fail(self):
 
612
        # the LockDir.lock_released hook does not fire on failure.
 
613
        ld = self.get_lock()
 
614
        ld.create()
 
615
        ld2 = self.get_lock()
 
616
        ld.attempt_lock()
 
617
        ld2.force_break(ld2.peek())
 
618
        LockDir.hooks.install_named_hook('lock_released',
 
619
                                         self.record_hook, 'record_hook')
 
620
        self.assertRaises(LockBroken, ld.unlock)
 
621
        self.assertEqual([], self._calls)
 
622
 
 
623
    def test_LockDir_broken_success(self):
 
624
        # the LockDir.lock_broken hook fires when a lock is broken.
 
625
        ld = self.get_lock()
 
626
        ld.create()
 
627
        ld2 = self.get_lock()
 
628
        result = ld.attempt_lock()
 
629
        LockDir.hooks.install_named_hook('lock_broken',
 
630
                                         self.record_hook, 'record_hook')
 
631
        ld2.force_break(ld2.peek())
 
632
        lock_path = ld.transport.abspath(ld.path)
 
633
        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)
 
634
 
 
635
    def test_LockDir_broken_failure(self):
 
636
        # the LockDir.lock_broken hook does not fires when a lock is already
 
637
        # released.
 
638
        ld = self.get_lock()
 
639
        ld.create()
 
640
        ld2 = self.get_lock()
 
641
        result = ld.attempt_lock()
 
642
        holder_info = ld2.peek()
 
643
        ld.unlock()
 
644
        LockDir.hooks.install_named_hook('lock_broken',
 
645
                                         self.record_hook, 'record_hook')
 
646
        ld2.force_break(holder_info)
 
647
        lock_path = ld.transport.abspath(ld.path)
 
648
        self.assertEqual([], self._calls)