~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2006-2010 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Test locks across all branch implemenations"""

from bzrlib import (
    branch as _mod_branch,
    errors,
    tests,
    )
from bzrlib.tests import (
    lock_helpers,
    per_branch,
    )
from bzrlib.tests.matchers import *


class TestBranchLocking(per_branch.TestCaseWithBranch):

    def setUp(self):
        super(TestBranchLocking, self).setUp()
        self.reduceLockdirTimeout()

    def get_instrumented_branch(self):
        """Get a Branch object which has been instrumented"""
        # TODO: jam 20060630 It may be that not all formats have a
        # 'control_files' member. So we should fail gracefully if
        # not there. But assuming it has them lets us test the exact
        # lock/unlock order.
        self.locks = []
        b = lock_helpers.LockWrapper(self.locks, self.get_branch(), 'b')
        b.repository = lock_helpers.LockWrapper(self.locks, b.repository, 'r')
        bcf = b.control_files
        rcf = getattr(b.repository, 'control_files', None)
        if rcf is None:
            self.combined_branch = False
        else:
            # Look out for branch types that reuse their control files
            self.combined_control = bcf is rcf
        try:
            b.control_files = lock_helpers.LockWrapper(
                self.locks, b.control_files, 'bc')
        except AttributeError:
            # RemoteBranch seems to trigger this.
            raise tests.TestSkipped(
                'Could not instrument branch control files.')
        if self.combined_control:
            # instrument the repository control files too to ensure its worked
            # with correctly. When they are not shared, we trust the repository
            # API and only instrument the repository itself.
            b.repository.control_files = lock_helpers.LockWrapper(
                self.locks, b.repository.control_files, 'rc')
        return b

    def test_01_lock_read(self):
        # Test that locking occurs in the correct order
        b = self.get_instrumented_branch()

        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())
        b.lock_read()
        try:
            self.assertTrue(b.is_locked())
            self.assertTrue(b.repository.is_locked())
        finally:
            b.unlock()
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())

        if self.combined_control:
            self.assertEqual([('b', 'lr', True),
                              ('r', 'lr', True),
                              ('rc', 'lr', True),
                              ('bc', 'lr', True),
                              ('b', 'ul', True),
                              ('bc', 'ul', True),
                              ('r', 'ul', True),
                              ('rc', 'ul', True),
                             ], self.locks)
        else:
            self.assertEqual([('b', 'lr', True),
                              ('r', 'lr', True),
                              ('bc', 'lr', True),
                              ('b', 'ul', True),
                              ('bc', 'ul', True),
                              ('r', 'ul', True),
                             ], self.locks)

    def test_02_lock_write(self):
        # Test that locking occurs in the correct order
        b = self.get_instrumented_branch()

        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())
        b.lock_write()
        try:
            self.assertTrue(b.is_locked())
            self.assertTrue(b.repository.is_locked())
        finally:
            b.unlock()
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())

        if self.combined_control:
            self.assertEqual([('b', 'lw', True),
                              ('r', 'lw', True),
                              ('rc', 'lw', True),
                              ('bc', 'lw', True),
                              ('b', 'ul', True),
                              ('bc', 'ul', True),
                              ('r', 'ul', True),
                              ('rc', 'ul', True),
                             ], self.locks)
        else:
            self.assertEqual([('b', 'lw', True),
                              ('r', 'lw', True),
                              ('bc', 'lw', True),
                              ('b', 'ul', True),
                              ('bc', 'ul', True),
                              ('r', 'ul', True),
                             ], self.locks)

    def test_03_lock_fail_unlock_repo(self):
        # Make sure branch.unlock() is called, even if there is a
        # failure while unlocking the repository.
        b = self.get_instrumented_branch()
        b.repository.disable_unlock()

        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())
        b.lock_write()
        try:
            self.assertTrue(b.is_locked())
            self.assertTrue(b.repository.is_locked())
            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)
            if self.combined_control:
                self.assertTrue(b.is_locked())
            else:
                self.assertFalse(b.is_locked())
            self.assertTrue(b.repository.is_locked())

            # We unlock the branch control files, even if
            # we fail to unlock the repository
            if self.combined_control:
                self.assertEqual([('b', 'lw', True),
                                  ('r', 'lw', True),
                                  ('rc', 'lw', True),
                                  ('bc', 'lw', True),
                                  ('b', 'ul', True),
                                  ('bc', 'ul', True),
                                  ('r', 'ul', False),
                                 ], self.locks)
            else:
                self.assertEqual([('b', 'lw', True),
                                  ('r', 'lw', True),
                                  ('bc', 'lw', True),
                                  ('b', 'ul', True),
                                  ('bc', 'ul', True),
                                  ('r', 'ul', False),
                                 ], self.locks)

        finally:
            # For cleanup purposes, make sure we are unlocked
            b.repository._other.unlock()

    def test_04_lock_fail_unlock_control(self):
        # Make sure repository.unlock() is not called, if we fail to unlock
        # self leaving ourselves still locked, so that attempts to recover
        # don't encounter an unlocked repository.
        b = self.get_instrumented_branch()
        b.control_files.disable_unlock()

        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())
        b.lock_write()
        try:
            self.assertTrue(b.is_locked())
            self.assertTrue(b.repository.is_locked())
            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)
            self.assertTrue(b.is_locked())
            self.assertTrue(b.repository.is_locked())

            # We unlock the repository even if
            # we fail to unlock the control files
            if self.combined_control:
                self.assertEqual([('b', 'lw', True),
                                  ('r', 'lw', True),
                                  ('rc', 'lw', True),
                                  ('bc', 'lw', True),
                                  ('b', 'ul', True),
                                  ('bc', 'ul', False),
                                  ('r', 'ul', True),
                                  ('rc', 'ul', True),
                                 ], self.locks)
            else:
                self.assertEqual([('b', 'lw', True),
                                  ('r', 'lw', True),
                                  ('bc', 'lw', True),
                                  ('b', 'ul', True),
                                  ('bc', 'ul', False),
                                 ], self.locks)

        finally:
            # For cleanup purposes, make sure we are unlocked
            b.control_files._other.unlock()

    def test_05_lock_read_fail_repo(self):
        # Test that the branch is not locked if it cannot lock the repository
        b = self.get_instrumented_branch()
        b.repository.disable_lock_read()

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())

        self.assertEqual([('b', 'lr', True),
                          ('r', 'lr', False),
                         ], self.locks)

    def test_06_lock_write_fail_repo(self):
        # Test that the branch is not locked if it cannot lock the repository
        b = self.get_instrumented_branch()
        b.repository.disable_lock_write()

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())

        self.assertEqual([('b', 'lw', True),
                          ('r', 'lw', False),
                         ], self.locks)

    def test_07_lock_read_fail_control(self):
        # Test the repository is unlocked if we can't lock self
        b = self.get_instrumented_branch()
        b.control_files.disable_lock_read()

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())

        if self.combined_control:
            self.assertEqual([('b', 'lr', True),
                              ('r', 'lr', True),
                              ('rc', 'lr', True),
                              ('bc', 'lr', False),
                              ('r', 'ul', True),
                              ('rc', 'ul', True),
                             ], self.locks)
        else:
            self.assertEqual([('b', 'lr', True),
                              ('r', 'lr', True),
                              ('bc', 'lr', False),
                              ('r', 'ul', True),
                             ], self.locks)

    def test_08_lock_write_fail_control(self):
        # Test the repository is unlocked if we can't lock self
        b = self.get_instrumented_branch()
        b.control_files.disable_lock_write()

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)
        self.assertFalse(b.is_locked())
        self.assertFalse(b.repository.is_locked())
        if self.combined_control:
            self.assertEqual([('b', 'lw', True),
                              ('r', 'lw', True),
                              ('rc', 'lw', True),
                              ('bc', 'lw', False),
                              ('r', 'ul', True),
                              ('rc', 'ul', True),
                             ], self.locks)
        else:
            self.assertEqual([('b', 'lw', True),
                              ('r', 'lw', True),
                              ('bc', 'lw', False),
                              ('r', 'ul', True),
                             ], self.locks)

    def test_lock_write_returns_None_refuses_token(self):
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is not None:
                # This test does not apply, because this lockable supports
                # tokens.
                return
            self.assertRaises(errors.TokenLockingNotSupported,
                              branch.lock_write, token='token')
        finally:
            branch.unlock()

    def test_reentering_lock_write_raises_on_token_mismatch(self):
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this lockable refuses
                # tokens.
                return
            different_branch_token = token + 'xxx'
            # Re-using the same lockable instance with a different branch token
            # will raise TokenMismatch.
            self.assertRaises(errors.TokenMismatch,
                              branch.lock_write,
                              token=different_branch_token)
        finally:
            branch.unlock()

    def test_lock_write_with_nonmatching_token(self):
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this branch refuses
                # tokens.
                return
            different_branch_token = token + 'xxx'

            new_branch = branch.bzrdir.open_branch()
            # We only want to test the relocking abilities of branch, so use the
            # existing repository object which is already locked.
            new_branch.repository = branch.repository
            self.assertRaises(errors.TokenMismatch,
                              new_branch.lock_write,
                              token=different_branch_token)
        finally:
            branch.unlock()


    def test_lock_write_with_matching_token(self):
        """Test that a branch can be locked with a token, if it is already
        locked by that token."""
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this branch refuses tokens.
                return
            # The same instance will accept a second lock_write if the specified
            # token matches.
            branch.lock_write(token=token)
            branch.unlock()
            # Calling lock_write on a new instance for the same lockable will
            # also succeed.
            new_branch = branch.bzrdir.open_branch()
            # We only want to test the relocking abilities of branch, so use the
            # existing repository object which is already locked.
            new_branch.repository = branch.repository
            new_branch.lock_write(token=token)
            new_branch.unlock()
        finally:
            branch.unlock()

    def test_unlock_after_lock_write_with_token(self):
        # If lock_write did not physically acquire the lock (because it was
        # passed some tokens), then unlock should not physically release it.
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this lockable refuses
                # tokens.
                return
            new_branch = branch.bzrdir.open_branch()
            # We only want to test the relocking abilities of branch, so use the
            # existing repository object which is already locked.
            new_branch.repository = branch.repository
            new_branch.lock_write(token=token)
            new_branch.unlock()
            self.assertTrue(branch.get_physical_lock_status()) #XXX
        finally:
            branch.unlock()

    def test_lock_write_with_token_fails_when_unlocked(self):
        # First, lock and then unlock to get superficially valid tokens.  This
        # mimics a likely programming error, where a caller accidentally tries
        # to lock with a token that is no longer valid (because the original
        # lock was released).
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        branch.unlock()
        if token is None:
            # This test does not apply, because this lockable refuses
            # tokens.
            return
        self.assertRaises(errors.TokenMismatch, branch.lock_write, token=token)

    def test_lock_write_reenter_with_token(self):
        branch = self.make_branch('b')
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this lockable refuses
                # tokens.
                return
            # Relock with a token.
            branch.lock_write(token=token)
            branch.unlock()
        finally:
            branch.unlock()
        # The lock should be unlocked on disk.  Verify that with a new lock
        # instance.
        new_branch = branch.bzrdir.open_branch()
        # Calling lock_write now should work, rather than raise LockContention.
        new_branch.lock_write()
        new_branch.unlock()

    def test_leave_lock_in_place(self):
        branch = self.make_branch('b')
        # Lock the branch, then use leave_lock_in_place so that when we
        # unlock the branch the lock is still held on disk.
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this repository refuses lock
                # tokens.
                self.assertRaises(NotImplementedError,
                                  branch.leave_lock_in_place)
                return
            branch.leave_lock_in_place()
        finally:
            branch.unlock()
        # We should be unable to relock the repo.
        self.assertRaises(errors.LockContention, branch.lock_write)
        # Cleanup
        branch.lock_write(token)
        branch.dont_leave_lock_in_place()
        branch.unlock()

    def test_dont_leave_lock_in_place(self):
        branch = self.make_branch('b')
        # Create a lock on disk.
        token = branch.lock_write().branch_token
        try:
            if token is None:
                # This test does not apply, because this branch refuses lock
                # tokens.
                self.assertRaises(NotImplementedError,
                                  branch.dont_leave_lock_in_place)
                return
            try:
                branch.leave_lock_in_place()
            except NotImplementedError:
                # This branch doesn't support this API.
                return
            try:
                branch.repository.leave_lock_in_place()
            except NotImplementedError:
                # This repo doesn't support leaving locks around,
                # assume it is essentially lock-free.
                repo_token = None
            else:
                repo_token = branch.repository.lock_write()
                branch.repository.unlock()
        finally:
            branch.unlock()
        # Reacquire the lock (with a different branch object) by using the
        # tokens.
        new_branch = branch.bzrdir.open_branch()
        if repo_token is not None:
            # We have to explicitly lock the repository first.
            new_branch.repository.lock_write(token=repo_token)
        new_branch.lock_write(token=token)
        if repo_token is not None:
            # Now we don't need our own repository lock anymore (the branch is
            # holding it for us).
            new_branch.repository.unlock()
        # Call dont_leave_lock_in_place, so that the lock will be released by
        # this instance, even though the lock wasn't originally acquired by it.
        new_branch.dont_leave_lock_in_place()
        if repo_token is not None:
            new_branch.repository.dont_leave_lock_in_place()
        new_branch.unlock()
        # Now the branch (and repository) is unlocked.  Test this by locking it
        # without tokens.
        branch.lock_write()
        branch.unlock()

    def test_lock_read_then_unlock(self):
        # Calling lock_read then unlocking should work without errors.
        branch = self.make_branch('b')
        branch.lock_read()
        branch.unlock()

    def test_lock_read_returns_unlockable(self):
        branch = self.make_branch('b')
        self.assertThat(branch.lock_read, ReturnsUnlockable(branch))

    def test_lock_write_locks_repo_too(self):
        if isinstance(self.branch_format, _mod_branch.BzrBranchFormat4):
            # Branch format 4 is combined with the repository, so this test
            # doesn't apply.
            return
        branch = self.make_branch('b')
        branch = branch.bzrdir.open_branch()
        branch.lock_write()
        try:
            # The branch should have asked the repository to lock.
            self.assertTrue(branch.repository.is_write_locked())
            # Does the repository type actually lock?
            if not branch.repository.get_physical_lock_status():
                # The test was successfully applied, so it was applicable.
                return
            # Now the branch.repository is physically locked, so we can't lock
            # it with a new repository instance.
            new_repo = branch.bzrdir.open_repository()
            self.assertRaises(errors.LockContention, new_repo.lock_write)
            # We can call lock_write on the original repository object though,
            # because it is already locked.
            branch.repository.lock_write()
            branch.repository.unlock()
        finally:
            branch.unlock()

    def test_lock_write_returns_unlockable(self):
        branch = self.make_branch('b')
        self.assertThat(branch.lock_write, ReturnsUnlockable(branch))

    def test_lock_write_raises_in_lock_read(self):
        branch = self.make_branch('b')
        branch.lock_read()
        self.addCleanup(branch.unlock)
        err = self.assertRaises(errors.ReadOnlyError, branch.lock_write)

    def test_lock_and_unlock_leaves_repo_unlocked(self):
        branch = self.make_branch('b')
        branch.lock_write()
        branch.unlock()
        self.assertRaises(errors.LockNotHeld, branch.repository.unlock)