← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

(vila) Make all transport put_bytes() raises TypeError when given unicode
 strings rather than bytes (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
17
17
"""Tests for LockDir"""
18
18
 
19
19
import os
20
 
import sys
21
20
import time
22
21
 
23
22
import bzrlib
44
43
from bzrlib.tests import (
45
44
    features,
46
45
    TestCase,
 
46
    TestCaseInTempDir,
47
47
    TestCaseWithTransport,
48
48
    )
49
 
from bzrlib.trace import note
50
49
 
51
50
# These tests are run on the default transport provided by the test framework
52
51
# (typically a local disk transport).  That can be changed by the --transport
54
53
# implementation are tested separately.  (The main requirement is just that
55
54
# they don't allow overwriting nonempty directories.)
56
55
 
 
56
 
57
57
class TestLockDir(TestCaseWithTransport):
58
58
    """Test LockDir operations"""
59
59
 
148
148
        # lock is held, should get some info on it
149
149
        info1 = lf1.peek()
150
150
        self.assertEqual(set(info1.info_dict.keys()),
151
 
                         set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
 
151
            set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
152
152
        # should get the same info if we look at it through a different
153
153
        # instance
154
154
        info2 = LockDir(t, 'test_lock').peek()
190
190
            # it should only take about 0.4 seconds, but we allow more time in
191
191
            # case the machine is heavily loaded
192
192
            self.assertTrue(after - before <= 8.0,
193
 
                    "took %f seconds to detect lock contention" % (after - before))
 
193
                "took %f seconds to detect lock contention" % (after - before))
194
194
        finally:
195
195
            lf1.unlock()
196
196
        self.assertEqual(1, len(self._logged_reports))
197
 
        self.assertEqual(self._logged_reports[0][0],
198
 
            '%s lock %s held by %s\n'
199
 
            'at %s [process #%s], acquired %s.\n'
200
 
            'Will continue to try until %s, unless '
201
 
            'you press Ctrl-C.\n'
202
 
            'See "bzr help break-lock" for more.')
203
 
        start, lock_url, user, hostname, pid, time_ago, deadline_str = \
204
 
            self._logged_reports[0][1]
205
 
        self.assertEqual(start, u'Unable to obtain')
206
 
        self.assertEqual(user, u'jrandom@example.com')
207
 
        # skip hostname
208
 
        self.assertContainsRe(pid, r'\d+')
209
 
        self.assertContainsRe(time_ago, r'.* ago')
210
 
        self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
 
197
        self.assertContainsRe(self._logged_reports[0][0],
 
198
            r'Unable to obtain lock .* held by jrandom@example\.com on .*'
 
199
            r' \(process #\d+\), acquired .* ago\.\n'
 
200
            r'Will continue to try until \d{2}:\d{2}:\d{2}, unless '
 
201
            r'you press Ctrl-C.\n'
 
202
            r'See "bzr help break-lock" for more.')
211
203
 
212
204
    def test_31_lock_wait_easy(self):
213
205
        """Succeed when waiting on a lock with no contention.
355
347
        ld.create()
356
348
        ld.lock_write()
357
349
        ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
 
350
 
358
351
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
359
352
            def __init__(self):
360
353
                self.prompts = []
 
354
 
361
355
            def get_boolean(self, prompt):
362
356
                self.prompts.append(('boolean', prompt))
363
357
                return True
 
358
 
364
359
        ui = LoggingUIFactory()
365
360
        self.overrideAttr(bzrlib.ui, 'ui_factory', ui)
366
361
        ld2.break_lock()
378
373
        ld.create()
379
374
        ld.lock_write()
380
375
        ld.transport.delete('test_lock/held/info')
 
376
 
381
377
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
382
378
            def __init__(self):
383
379
                self.prompts = []
 
380
 
384
381
            def get_boolean(self, prompt):
385
382
                self.prompts.append(('boolean', prompt))
386
383
                return True
 
384
 
387
385
        ui = LoggingUIFactory()
388
386
        orig_factory = bzrlib.ui.ui_factory
389
387
        bzrlib.ui.ui_factory = ui
427
425
        ld1.create()
428
426
        ld1.lock_write()
429
427
        try:
430
 
            info_list = ld1.peek().to_readable_list()
 
428
            info_list = ld1.peek().to_readable_dict()
431
429
        finally:
432
430
            ld1.unlock()
433
 
        self.assertEqual(info_list[0], u'jrandom@example.com')
434
 
        # info_list[1] is hostname. we skip this.
435
 
        self.assertContainsRe(info_list[2], '^\d+$') # pid
436
 
        self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
 
431
        self.assertEqual(info_list['user'], u'jrandom@example.com')
 
432
        self.assertContainsRe(info_list['pid'], '^\d+$')
 
433
        self.assertContainsRe(info_list['time_ago'], r'^\d+ seconds? ago$')
437
434
 
438
435
    def test_lock_without_email(self):
439
 
        global_config = config.GlobalConfig()
 
436
        global_config = config.GlobalStack()
440
437
        # Intentionally has no email address
441
 
        global_config.set_user_option('email', 'User Identity')
 
438
        global_config.set('email', 'User Identity')
442
439
        ld1 = self.get_lock()
443
440
        ld1.create()
444
441
        ld1.lock_write()
467
464
    def test_lock_with_buggy_rename(self):
468
465
        # test that lock acquisition handles servers which pretend they
469
466
        # renamed correctly but that actually fail
470
 
        t = transport.get_transport('brokenrename+' + self.get_url())
 
467
        t = transport.get_transport_from_url(
 
468
            'brokenrename+' + self.get_url())
471
469
        ld1 = LockDir(t, 'test_lock')
472
470
        ld1.create()
473
471
        ld1.attempt_lock()
487
485
        # should be nothing before we start
488
486
        ld1.create()
489
487
        t = self.get_transport().clone('test_lock')
 
488
 
490
489
        def check_dir(a):
491
490
            self.assertEquals(a, t.list_dir('.'))
 
491
 
492
492
        check_dir([])
493
493
        # when held, that's all we see
494
494
        ld1.attempt_lock()
511
511
        t.put_bytes('test_lock/held/info', '')
512
512
        lf = LockDir(t, 'test_lock')
513
513
        info = lf.peek()
514
 
        formatted_info = info.to_readable_list()
 
514
        formatted_info = info.to_readable_dict()
515
515
        self.assertEquals(
516
 
            ['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
 
516
            dict(user='<unknown>', hostname='<unknown>', pid='<unknown>',
 
517
                time_ago='(unknown)'),
517
518
            formatted_info)
518
519
 
519
520
    def test_corrupt_lockdir_info(self):
654
655
        self.assertEqual([], self._calls)
655
656
 
656
657
 
657
 
class TestLockHeldInfo(TestCase):
 
658
class TestLockHeldInfo(TestCaseInTempDir):
658
659
    """Can get information about the lock holder, and detect whether they're
659
660
    still alive."""
660
661
 
 
662
    def test_repr(self):
 
663
        info = LockHeldInfo.for_this_process(None)
 
664
        self.assertContainsRe(repr(info), r"LockHeldInfo\(.*\)")
 
665
 
 
666
    def test_unicode(self):
 
667
        info = LockHeldInfo.for_this_process(None)
 
668
        self.assertContainsRe(unicode(info),
 
669
            r'held by .* on .* \(process #\d+\), acquired .* ago')
 
670
 
661
671
    def test_is_locked_by_this_process(self):
662
672
        info = LockHeldInfo.for_this_process(None)
663
673
        self.assertTrue(info.is_locked_by_this_process())
664
674
 
665
675
    def test_is_not_locked_by_this_process(self):
666
676
        info = LockHeldInfo.for_this_process(None)
667
 
        info.info_dict['pid'] = '123123123123123' # probably not us
 
677
        info.info_dict['pid'] = '123123123123123'
668
678
        self.assertFalse(info.is_locked_by_this_process())
669
679
 
670
680
    def test_lock_holder_live_process(self):
671
681
        """Detect that the holder (this process) is still running."""
672
682
        info = LockHeldInfo.for_this_process(None)
673
683
        self.assertFalse(info.is_lock_holder_known_dead())
674
 
        
 
684
 
675
685
    def test_lock_holder_dead_process(self):
676
686
        """Detect that the holder (this process) is still running."""
 
687
        self.overrideAttr(lockdir, 'get_host_name',
 
688
            lambda: 'aproperhostname')
677
689
        info = LockHeldInfo.for_this_process(None)
678
 
        info.info_dict['pid'] = '123123123' # probably not alive at all
679
 
        if sys.platform == 'win32':
680
 
            self.knownFailure(
681
 
                'live lock holder detection not implemented yet on win32')
 
690
        info.info_dict['pid'] = '123123123'
682
691
        self.assertTrue(info.is_lock_holder_known_dead())
683
 
        
 
692
 
684
693
    def test_lock_holder_other_machine(self):
685
694
        """The lock holder isn't here so we don't know if they're alive."""
686
695
        info = LockHeldInfo.for_this_process(None)
687
 
        info.info_dict['host'] = 'egg.example.com'
 
696
        info.info_dict['hostname'] = 'egg.example.com'
 
697
        info.info_dict['pid'] = '123123123'
 
698
        self.assertFalse(info.is_lock_holder_known_dead())
 
699
 
 
700
    def test_lock_holder_other_user(self):
 
701
        """Only auto-break locks held by this user."""
 
702
        info = LockHeldInfo.for_this_process(None)
 
703
        info.info_dict['user'] = 'notme@example.com'
 
704
        info.info_dict['pid'] = '123123123'
688
705
        self.assertFalse(info.is_lock_holder_known_dead())
689
706
 
690
707
    def test_no_good_hostname(self):
707
724
 
708
725
    :see: https://bugs.launchpad.net/bzr/+bug/220464
709
726
    """
 
727
 
710
728
    def test_auto_break_stale_lock(self):
711
729
        """Locks safely known to be stale are just cleaned up.
712
730
 
713
731
        This generates a warning but no other user interaction.
714
732
        """
 
733
        self.overrideAttr(lockdir, 'get_host_name',
 
734
            lambda: 'aproperhostname')
 
735
        # This is off by default at present; see the discussion in the bug.
 
736
        # If you change the default, don't forget to update the docs.
 
737
        config.GlobalStack().set('locks.steal_dead', True)
715
738
        # Create a lock pretending to come from a different nonexistent
716
739
        # process on the same machine.
717
740
        l1 = LockDir(self.get_transport(), 'a',
731
754
        token_1 = l1.attempt_lock()
732
755
        self.addCleanup(l1.unlock)
733
756
        l2 = LockDir(self.get_transport(), 'a')
734
 
        # This fails now, because dead lock breaking is turned off.
735
 
        config.GlobalConfig().set_user_option('steal_dead_locks', False)
 
757
        # This fails now, because dead lock breaking is off by default.
736
758
        self.assertRaises(LockContention,
737
759
            l2.attempt_lock)
738
760
        # and it's in fact not broken