← 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: Jelmer Vernooij
  • Date: 2016-04-03 16:32:31 UTC
  • mto: This revision was merged to the branch mainline in revision 6617.
  • Revision ID: jelmer@jelmer.uk-20160403163231-h72bo0uyek2gikw0
https://launchpad.net/bugs/1565503
Don't put French text in doc/en/user-reference when LANGUAGE=fr_CH.UTF_8.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006-2011 Canonical Ltd
 
1
# Copyright (C) 2006-2012, 2016 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
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
434
433
        self.assertContainsRe(info_list['time_ago'], r'^\d+ seconds? ago$')
435
434
 
436
435
    def test_lock_without_email(self):
437
 
        global_config = config.GlobalConfig()
 
436
        global_config = config.GlobalStack()
438
437
        # Intentionally has no email address
439
 
        global_config.set_user_option('email', 'User Identity')
 
438
        global_config.set('email', 'User Identity')
440
439
        ld1 = self.get_lock()
441
440
        ld1.create()
442
441
        ld1.lock_write()
465
464
    def test_lock_with_buggy_rename(self):
466
465
        # test that lock acquisition handles servers which pretend they
467
466
        # renamed correctly but that actually fail
468
 
        t = transport.get_transport('brokenrename+' + self.get_url())
 
467
        t = transport.get_transport_from_url(
 
468
            'brokenrename+' + self.get_url())
469
469
        ld1 = LockDir(t, 'test_lock')
470
470
        ld1.create()
471
471
        ld1.attempt_lock()
475
475
        # now the original caller should succeed in unlocking
476
476
        ld1.unlock()
477
477
        # and there should be nothing left over
478
 
        self.assertEquals([], t.list_dir('test_lock'))
 
478
        self.assertEqual([], t.list_dir('test_lock'))
479
479
 
480
480
    def test_failed_lock_leaves_no_trash(self):
481
481
        # if we fail to acquire the lock, we don't leave pending directories
487
487
        t = self.get_transport().clone('test_lock')
488
488
 
489
489
        def check_dir(a):
490
 
            self.assertEquals(a, t.list_dir('.'))
 
490
            self.assertEqual(a, t.list_dir('.'))
491
491
 
492
492
        check_dir([])
493
493
        # when held, that's all we see
512
512
        lf = LockDir(t, 'test_lock')
513
513
        info = lf.peek()
514
514
        formatted_info = info.to_readable_dict()
515
 
        self.assertEquals(
 
515
        self.assertEqual(
516
516
            dict(user='<unknown>', hostname='<unknown>', pid='<unknown>',
517
517
                time_ago='(unknown)'),
518
518
            formatted_info)
655
655
        self.assertEqual([], self._calls)
656
656
 
657
657
 
658
 
class TestLockHeldInfo(TestCase):
 
658
class TestLockHeldInfo(TestCaseInTempDir):
659
659
    """Can get information about the lock holder, and detect whether they're
660
660
    still alive."""
661
661
 
684
684
 
685
685
    def test_lock_holder_dead_process(self):
686
686
        """Detect that the holder (this process) is still running."""
 
687
        self.overrideAttr(lockdir, 'get_host_name',
 
688
            lambda: 'aproperhostname')
687
689
        info = LockHeldInfo.for_this_process(None)
688
690
        info.info_dict['pid'] = '123123123'
689
 
        if sys.platform == 'win32':
690
 
            self.knownFailure(
691
 
                'live lock holder detection not implemented yet on win32')
692
691
        self.assertTrue(info.is_lock_holder_known_dead())
693
692
 
694
693
    def test_lock_holder_other_machine(self):
731
730
 
732
731
        This generates a warning but no other user interaction.
733
732
        """
 
733
        self.overrideAttr(lockdir, 'get_host_name',
 
734
            lambda: 'aproperhostname')
734
735
        # This is off by default at present; see the discussion in the bug.
735
736
        # If you change the default, don't forget to update the docs.
736
 
        config.GlobalConfig().set_user_option('locks.steal_dead', True)
 
737
        config.GlobalStack().set('locks.steal_dead', True)
737
738
        # Create a lock pretending to come from a different nonexistent
738
739
        # process on the same machine.
739
740
        l1 = LockDir(self.get_transport(), 'a',