~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2007-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

"""Tests for Branch.revision_history and last_revision."""

from bzrlib import (

    branch,

    errors,

    revision as _mod_revision,

    )

from bzrlib.symbol_versioning import deprecated_in

from bzrlib.tests import (

    per_branch,

    TestNotApplicable,

    )

class TestLastRevision(per_branch.TestCaseWithBranch):

    """Tests for the last_revision property of the branch.

    """

    def test_set_last_revision_info(self):

        # based on TestBranch.test_append_revisions, which uses the old

        # append_revision api

        wt = self.make_branch_and_tree('tree')

        wt.commit('f', rev_id='rev1')

        wt.commit('f', rev_id='rev2')

        wt.commit('f', rev_id='rev3')

        br = self.get_branch()

        br.fetch(wt.branch)

        br.set_last_revision_info(1, 'rev1')

        self.assertEquals(br.revision_history(), ["rev1",])

        br.set_last_revision_info(3, 'rev3')

        self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])

        # append_revision specifically prohibits some ids;

        # set_last_revision_info currently does not

        ## self.assertRaises(errors.ReservedId,

        ##         br.set_last_revision_info, 4, 'current:')

class TestRevisionHistoryCaching(per_branch.TestCaseWithBranch):

    """Tests for the caching of branches' revision_history.

    When locked, branches should avoid regenerating or rereading

    revision_history by caching the last value of it.  This is safe because

    the branch is locked, so nothing can change the revision_history

    unexpectedly.

    When not locked, obviously the revision_history will need to be regenerated

    or reread each time.

    We test if revision_history is using the cache by instrumenting the branch's

    _gen_revision_history method, which is called by Branch.revision_history if

    the branch does not have a cache of the revision history.

    """

    def get_instrumented_branch(self):

        """Get a branch and monkey patch it to log calls to

        _gen_revision_history.

        :returns: a tuple of (the branch, list that calls will be logged to)

        """

        branch = self.get_branch()

        calls = []

        real_gen_revision_history = branch._gen_revision_history

        def fake_gen_revision_history():

            calls.append('_gen_revision_history')

            return real_gen_revision_history()

        branch._gen_revision_history = fake_gen_revision_history

        return branch, calls

    def test_revision_history_when_unlocked(self):

        """Repeated calls to revision history will call _gen_revision_history

        each time when the branch is not locked.

        """

        branch, calls = self.get_instrumented_branch()

        # Repeatedly call revision_history.

        branch.revision_history()

        branch.revision_history()

        self.assertEqual(

            ['_gen_revision_history', '_gen_revision_history'], calls)

    def test_revision_history_when_locked(self):

        """Repeated calls to revision history will only call

        _gen_revision_history once while the branch is locked.

        """

        branch, calls = self.get_instrumented_branch()

        # Lock the branch, then repeatedly call revision_history.

        branch.lock_read()

        try:

            branch.revision_history()

            branch.revision_history()

            self.assertEqual(['_gen_revision_history'], calls)

        finally:

            branch.unlock()

    def test_set_revision_history_when_locked(self):

        """When the branch is locked, calling set_revision_history should cache

        the revision history so that a later call to revision_history will not

        need to call _gen_revision_history.

        """

        branch, calls = self.get_instrumented_branch()

        # Lock the branch, set the revision history, then repeatedly call

        # revision_history.

        branch.lock_write()

        self.applyDeprecated(deprecated_in((2, 4, 0)),

            branch.set_revision_history, [])

        try:

            branch.revision_history()

            self.assertEqual([], calls)

        finally:

            branch.unlock()

    def test_set_revision_history_when_unlocked(self):

        """When the branch is not locked, calling set_revision_history will not

        cause the revision history to be cached.

        """

        branch, calls = self.get_instrumented_branch()

        self.applyDeprecated(deprecated_in((2, 4, 0)),

            branch.set_revision_history, [])

        branch.revision_history()

        self.assertEqual(['_gen_revision_history'], calls)

    def test_set_last_revision_info_when_locked(self):

        """When the branch is locked, calling set_last_revision_info should

        cache the last revision info so that a later call to last_revision_info

        will not need the revision_history.  Thus the branch will not to call

        _gen_revision_history in this situation.

        """

        a_branch, calls = self.get_instrumented_branch()

        # Lock the branch, set the last revision info, then call

        # last_revision_info.

        a_branch.lock_write()

        a_branch.set_last_revision_info(0, _mod_revision.NULL_REVISION)

        del calls[:]

        try:

            a_branch.last_revision_info()

            self.assertEqual([], calls)

        finally:

            a_branch.unlock()

    def test_set_last_revision_info_none(self):

        """Passing None to set_last_revision_info raises an exception."""

        a_branch = self.get_branch()

        # Lock the branch, set the last revision info, then call

        # last_revision_info.

        a_branch.lock_write()

        self.addCleanup(a_branch.unlock)

        self.assertRaises(errors.InvalidRevisionId,

            a_branch.set_last_revision_info, 0, None)

    def test_set_last_revision_info_uncaches_revision_history_for_format6(self):

        """On format 6 branches, set_last_revision_info invalidates the revision

        history cache.

        """

        if not isinstance(self.branch_format, branch.BzrBranchFormat6):

            return

        a_branch, calls = self.get_instrumented_branch()

        # Lock the branch, cache the revision history.

        a_branch.lock_write()

        a_branch.revision_history()

        # Set the last revision info, clearing the cache.

        a_branch.set_last_revision_info(0, _mod_revision.NULL_REVISION)

        del calls[:]

        try:

            a_branch.revision_history()

            self.assertEqual(['_gen_revision_history'], calls)

        finally:

            a_branch.unlock()

    def test_cached_revision_history_not_accidentally_mutable(self):

        """When there's a cached version of the history, revision_history

        returns a copy of the cached data so that callers cannot accidentally

        corrupt the cache.

        """

        branch = self.get_branch()

        # Lock the branch, then repeatedly call revision_history, mutating the

        # results.

        branch.lock_read()

        try:

            # The first time the data returned will not be in the cache.

            history = branch.revision_history()

            history.append('one')

            # The second time the data comes from the cache.

            history = branch.revision_history()

            history.append('two')

            # The revision_history() should still be unchanged, even though

            # we've mutated the return values from earlier calls.

            self.assertEqual([], branch.revision_history())

        finally:

            branch.unlock()

class TestRevisionHistory(per_branch.TestCaseWithBranch):

    def test_parent_ghost(self):

        tree = self.make_branch_and_tree('tree')

        if not tree.branch.repository._format.supports_ghosts:

            raise TestNotApplicable("repository format does not support ghosts")

        tree.add_parent_tree_id('ghost-revision',

                                allow_leftmost_as_ghost=True)

        tree.commit('first non-ghost commit', rev_id='non-ghost-revision')

        self.assertEqual(['non-ghost-revision'],

                         tree.branch.revision_history())