/brz/remove-bazaar

# Copyright (C) 2006-2011 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 repository commit builder."""

import os

from bzrlib import (

    config,

    errors,

    inventory,

    osutils,

    repository,

    revision as _mod_revision,

    tests,

    )

from bzrlib.tests import per_repository

from bzrlib.tests import (

    features,

    )

class TestCommitBuilder(per_repository.TestCaseWithRepository):

    def test_get_commit_builder(self):

        branch = self.make_branch('.')

        branch.repository.lock_write()

        builder = branch.repository.get_commit_builder(

            branch, [], branch.get_config_stack())

        self.assertIsInstance(builder, repository.CommitBuilder)

        self.assertTrue(builder.random_revid)

        branch.repository.commit_write_group()

        branch.repository.unlock()

    def record_root(self, builder, tree):

        if builder.record_root_entry is True:

            tree.lock_read()

            try:

                ie = tree.inventory.root

            finally:

                tree.unlock()

            parent_tree = tree.branch.repository.revision_tree(

                              _mod_revision.NULL_REVISION)

            parent_invs = []

            builder.record_entry_contents(ie, parent_invs, '', tree,

                tree.path_content_summary(''))

    def test_finish_inventory_with_record_root(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't support "

                    "record_entry_contents")

            repo = tree.branch.repository

            self.record_root(builder, tree)

            builder.finish_inventory()

            repo.commit_write_group()

        finally:

            tree.unlock()

    def test_finish_inventory_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            try:

                list(builder.record_iter_changes(tree, tree.last_revision(),

                    tree.iter_changes(tree.basis_tree())))

                builder.finish_inventory()

            except:

                builder.abort()

                raise

            repo = tree.branch.repository

            repo.commit_write_group()

        finally:

            tree.unlock()

    def test_abort_record_entry_contents(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't support "

                    "record_entry_contents")

            self.record_root(builder, tree)

            builder.finish_inventory()

            builder.abort()

        finally:

            tree.unlock()

    def test_abort_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            try:

                basis = tree.basis_tree()

                last_rev = tree.last_revision()

                changes = tree.iter_changes(basis)

                list(builder.record_iter_changes(tree, last_rev, changes))

                builder.finish_inventory()

            finally:

                builder.abort()

        finally:

            tree.unlock()

    def test_commit_lossy(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([], lossy=True)

            list(builder.record_iter_changes(tree, tree.last_revision(),

                tree.iter_changes(tree.basis_tree())))

            builder.finish_inventory()

            rev_id = builder.commit('foo bar blah')

        finally:

            tree.unlock()

        rev = tree.branch.repository.get_revision(rev_id)

        self.assertEqual('foo bar blah', rev.message)

    def test_commit_message(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            list(builder.record_iter_changes(tree, tree.last_revision(),

                tree.iter_changes(tree.basis_tree())))

            builder.finish_inventory()

            rev_id = builder.commit('foo bar blah')

        finally:

            tree.unlock()

        rev = tree.branch.repository.get_revision(rev_id)

        self.assertEqual('foo bar blah', rev.message)

    def test_updates_branch(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            list(builder.record_iter_changes(tree, tree.last_revision(),

                tree.iter_changes(tree.basis_tree())))

            builder.finish_inventory()

            will_update_branch = builder.updates_branch

            rev_id = builder.commit('might update the branch')

        finally:

            tree.unlock()

        actually_updated_branch = (tree.branch.last_revision() == rev_id)

        self.assertEquals(actually_updated_branch, will_update_branch)

    def test_commit_with_revision_id_record_entry_contents(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            # use a unicode revision id to test more corner cases.

            # The repository layer is meant to handle this.

            revision_id = u'\xc8abc'.encode('utf8')

            try:

                try:

                    builder = tree.branch.get_commit_builder([],

                        revision_id=revision_id)

                except errors.NonAsciiRevisionId:

                    revision_id = 'abc'

                    builder = tree.branch.get_commit_builder([],

                        revision_id=revision_id)

            except errors.CannotSetRevisionId:

                # This format doesn't support supplied revision ids

                return

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't support "

                    "record_entry_contents")

            self.assertFalse(builder.random_revid)

            self.record_root(builder, tree)

            builder.finish_inventory()

            self.assertEqual(revision_id, builder.commit('foo bar'))

        finally:

            tree.unlock()

        self.assertTrue(tree.branch.repository.has_revision(revision_id))

        # the revision id must be set on the inventory when saving it. This

        # does not precisely test that - a repository that wants to can add it

        # on deserialisation, but thats all the current contract guarantees

        # anyway.

        self.assertEqual(revision_id,

            tree.branch.repository.get_inventory(revision_id).revision_id)

    def test_commit_with_revision_id_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            # use a unicode revision id to test more corner cases.

            # The repository layer is meant to handle this.

            revision_id = u'\xc8abc'.encode('utf8')

            try:

                try:

                    builder = tree.branch.get_commit_builder([],

                        revision_id=revision_id)

                except errors.NonAsciiRevisionId:

                    revision_id = 'abc'

                    builder = tree.branch.get_commit_builder([],

                        revision_id=revision_id)

            except errors.CannotSetRevisionId:

                # This format doesn't support supplied revision ids

                return

            self.assertFalse(builder.random_revid)

            try:

                list(builder.record_iter_changes(tree, tree.last_revision(),

                    tree.iter_changes(tree.basis_tree())))

                builder.finish_inventory()

            except:

                builder.abort()

                raise

            self.assertEqual(revision_id, builder.commit('foo bar'))

        finally:

            tree.unlock()

        self.assertTrue(tree.branch.repository.has_revision(revision_id))

        # the revision id must be set on the inventory when saving it. This

        # does not precisely test that - a repository that wants to can add it

        # on deserialisation, but thats all the current contract guarantees

        # anyway.

        self.assertEqual(revision_id,

            tree.branch.repository.revision_tree(revision_id).get_revision_id())

    def test_commit_without_root_errors(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            def do_commit():

                try:

                    list(builder.record_iter_changes(

                        tree, tree.last_revision(), []))

                    builder.finish_inventory()

                except:

                    builder.abort()

                    raise

                else:

                    builder.commit("msg")

            self.assertRaises(errors.RootMissing, do_commit)

        finally:

            tree.unlock()

    def test_commit_without_root_or_record_iter_changes_errors(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            self.build_tree(['foo'])

            tree.add('foo', 'foo-id')

            builder = tree.branch.get_commit_builder([])

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't support "

                    "record_entry_contents")

            entry = tree.inventory['foo-id']

            self.assertRaises(errors.RootMissing,

                builder.record_entry_contents, entry, [], 'foo', tree,

                    tree.path_content_summary('foo'))

            builder.abort()

        finally:

            tree.unlock()

    def test_commit_unchanged_root_record_entry_contents(self):

        tree = self.make_branch_and_tree(".")

        old_revision_id = tree.commit('')

        tree.lock_write()

        parent_tree = tree.basis_tree()

        parent_tree.lock_read()

        self.addCleanup(parent_tree.unlock)

        builder = tree.branch.get_commit_builder([old_revision_id])

        try:

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't support "

                    "record_entry_contents")

            ie = inventory.make_entry('directory', '', None,

                    tree.get_root_id())

            delta, version_recorded, fs_hash = builder.record_entry_contents(

                ie, [parent_tree.inventory], '', tree,

                tree.path_content_summary(''))

            # Regardless of repository root behaviour we should consider this a

            # pointless commit.

            self.assertFalse(builder.any_changes())

            self.assertFalse(version_recorded)

            # if the repository format recorded a new root revision, that

            # should be in the delta

            got_new_revision = ie.revision != old_revision_id

            if got_new_revision:

                self.assertEqual(('', '', ie.file_id, ie), delta)

                # The delta should be tracked

                self.assertEqual(delta, builder._basis_delta[-1])

            else:

                self.assertEqual(None, delta)

            # Directories do not get hashed.

            self.assertEqual(None, fs_hash)

            builder.abort()

        except:

            builder.abort()

            tree.unlock()

            raise

        else:

            tree.unlock()

    def test_commit_unchanged_root_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        old_revision_id = tree.commit('')

        tree.lock_write()

        builder = tree.branch.get_commit_builder([old_revision_id])

        try:

            list(builder.record_iter_changes(tree, old_revision_id, []))

            # Regardless of repository root behaviour we should consider this a

            # pointless commit.

            self.assertFalse(builder.any_changes())

            builder.finish_inventory()

            builder.commit('')

            builder_tree = builder.revision_tree()

            new_root_id = builder_tree.get_root_id()

            new_root_revision = builder_tree.get_file_revision(new_root_id)

            if tree.branch.repository.supports_rich_root():

                # We should not have seen a new root revision

                self.assertEqual(old_revision_id, new_root_revision)

            else:

                # We should see a new root revision

                self.assertNotEqual(old_revision_id, new_root_revision)

        finally:

            tree.unlock()

    def test_commit_record_entry_contents(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't "

                    "support record_entry_contents")

            self.record_root(builder, tree)

            builder.finish_inventory()

            rev_id = builder.commit('foo bar')

        finally:

            tree.unlock()

        self.assertNotEqual(None, rev_id)

        self.assertTrue(tree.branch.repository.has_revision(rev_id))

        # the revision id must be set on the inventory when saving it. This does not

        # precisely test that - a repository that wants to can add it on deserialisation,

        # but thats all the current contract guarantees anyway.

        self.assertEqual(rev_id, tree.branch.repository.get_inventory(rev_id).revision_id)

    def test_get_basis_delta(self):

        tree = self.make_branch_and_tree(".")

        self.build_tree(["foo"])

        tree.add(["foo"], ["foo-id"])

        old_revision_id = tree.commit("added foo")

        tree.lock_write()

        try:

            self.build_tree(['bar'])

            tree.add(['bar'], ['bar-id'])

            basis = tree.branch.repository.revision_tree(old_revision_id)

            basis.lock_read()

            self.addCleanup(basis.unlock)

            builder = tree.branch.get_commit_builder([old_revision_id])

            total_delta = []

            try:

                if not builder.supports_record_entry_contents:

                    raise tests.TestNotApplicable("CommitBuilder doesn't "

                        "support record_entry_contents")

                parent_invs = [basis.inventory]

                builder.will_record_deletes()

                if builder.record_root_entry:

                    ie = basis.inventory.root.copy()

                    delta, _, _ = builder.record_entry_contents(ie, parent_invs,

                        '', tree, tree.path_content_summary(''))

                    if delta is not None:

                        total_delta.append(delta)

                delta = builder.record_delete("foo", "foo-id")

                total_delta.append(delta)

                new_bar = inventory.make_entry('file', 'bar',

                    parent_id=tree.get_root_id(), file_id='bar-id')

                delta, _, _ = builder.record_entry_contents(new_bar, parent_invs,

                    'bar', tree, tree.path_content_summary('bar'))

                total_delta.append(delta)

                # All actions should have been recorded in the basis_delta

                self.assertEqual(total_delta, builder.get_basis_delta())

                builder.finish_inventory()

                builder.commit('delete foo, add bar')

            except:

                tree.branch.repository.abort_write_group()

                raise

        finally:

            tree.unlock()

    def test_get_basis_delta_without_notification(self):

        tree = self.make_branch_and_tree(".")

        old_revision_id = tree.commit('')

        tree.lock_write()

        try:

            parent_tree = tree.basis_tree()

            parent_tree.lock_read()

            self.addCleanup(parent_tree.unlock)

            builder = tree.branch.get_commit_builder([old_revision_id])

            # It is an error to expect builder.get_basis_delta() to be correct,

            # if you have not also called will_record_deletes() to indicate you

            # will be calling record_delete() when appropriate

            self.assertRaises(AssertionError, builder.get_basis_delta)

            tree.branch.repository.abort_write_group()

        finally:

            tree.unlock()

    def test_record_delete(self):

        tree = self.make_branch_and_tree(".")

        self.build_tree(["foo"])

        tree.add(["foo"], ["foo-id"])

        rev_id = tree.commit("added foo")

        # Remove the inventory details for foo-id, because

        # record_entry_contents ends up copying root verbatim.

        tree.unversion(["foo-id"])

        tree.lock_write()

        try:

            basis = tree.branch.repository.revision_tree(rev_id)

            builder = tree.branch.get_commit_builder([rev_id])

            try:

                if not builder.supports_record_entry_contents:

                    raise tests.TestNotApplicable("CommitBuilder doesn't "

                        "support record_entry_contents")

                builder.will_record_deletes()

                if builder.record_root_entry is True:

                    parent_invs = [basis.inventory]

                    del basis.inventory.root.children['foo']

                    builder.record_entry_contents(basis.inventory.root,

                        parent_invs, '', tree, tree.path_content_summary(''))

                # the delta should be returned, and recorded in _basis_delta

                delta = builder.record_delete("foo", "foo-id")

                self.assertEqual(("foo", None, "foo-id", None), delta)

                self.assertEqual(delta, builder._basis_delta[-1])

                builder.finish_inventory()

                rev_id2 = builder.commit('delete foo')

            except:

                tree.branch.repository.abort_write_group()

                raise

        finally:

            tree.unlock()

        rev_tree = builder.revision_tree()

        rev_tree.lock_read()

        self.addCleanup(rev_tree.unlock)

        self.assertFalse(rev_tree.path2id('foo'))

    def test_record_delete_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        self.build_tree(["foo"])

        tree.add(["foo"], ["foo-id"])

        rev_id = tree.commit("added foo")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([rev_id])

            try:

                delete_change = ('foo-id', ('foo', None), True, (True, False),

                    (tree.path2id(''), None), ('foo', None), ('file', None),

                    (False, None))

                list(builder.record_iter_changes(tree, rev_id,

                    [delete_change]))

                self.assertEqual(("foo", None, "foo-id", None),

                    builder._basis_delta[0])

                self.assertTrue(builder.any_changes())

                builder.finish_inventory()

                rev_id2 = builder.commit('delete foo')

            except:

                builder.abort()

                raise

        finally:

            tree.unlock()

        rev_tree = builder.revision_tree()

        rev_tree.lock_read()

        self.addCleanup(rev_tree.unlock)

        self.assertFalse(rev_tree.path2id('foo'))

    def test_record_delete_without_notification(self):

        tree = self.make_branch_and_tree(".")

        self.build_tree(["foo"])

        tree.add(["foo"], ["foo-id"])

        rev_id = tree.commit("added foo")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([rev_id])

            try:

                if not builder.supports_record_entry_contents:

                    raise tests.TestNotApplicable("CommitBuilder doesn't "

                        "support record_entry_contents")

                self.record_root(builder, tree)

                self.assertRaises(AssertionError,

                    builder.record_delete, "foo", "foo-id")

            finally:

                tree.branch.repository.abort_write_group()

        finally:

            tree.unlock()

    def test_revision_tree_record_entry_contents(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            if not builder.supports_record_entry_contents:

                raise tests.TestNotApplicable("CommitBuilder doesn't "

                    "support record_entry_contents")

            self.record_root(builder, tree)

            builder.finish_inventory()

            rev_id = builder.commit('foo bar')

        finally:

            tree.unlock()

        rev_tree = builder.revision_tree()

        # Just a couple simple tests to ensure that it actually follows

        # the RevisionTree api.

        self.assertEqual(rev_id, rev_tree.get_revision_id())

        self.assertEqual([], rev_tree.get_parent_ids())

    def test_revision_tree_record_iter_changes(self):

        tree = self.make_branch_and_tree(".")

        tree.lock_write()

        try:

            builder = tree.branch.get_commit_builder([])

            try:

                list(builder.record_iter_changes(tree,

                    _mod_revision.NULL_REVISION,

                    tree.iter_changes(tree.basis_tree())))

                builder.finish_inventory()

                rev_id = builder.commit('foo bar')

            except:

                builder.abort()

                raise

            rev_tree = builder.revision_tree()

            # Just a couple simple tests to ensure that it actually follows

            # the RevisionTree api.

            self.assertEqual(rev_id, rev_tree.get_revision_id())

            self.assertEqual((), tuple(rev_tree.get_parent_ids()))

        finally:

            tree.unlock()

    def test_root_entry_has_revision(self):

        # test the root revision created and put in the basis

        # has the right rev id.

        # XXX: RBC 20081118 - this test is too big, it depends on the exact

        # behaviour of tree methods and so on; it should be written to the

        # commit builder interface directly.

        tree = self.make_branch_and_tree('.')

        rev_id = tree.commit('message')

        basis_tree = tree.basis_tree()

        basis_tree.lock_read()

        self.addCleanup(basis_tree.unlock)

        self.assertEqual(rev_id,

            basis_tree.get_file_revision(basis_tree.get_root_id()))

    def _get_revtrees(self, tree, revision_ids):

        tree.lock_read()

        try:

            trees = list(tree.branch.repository.revision_trees(revision_ids))

            for _tree in trees:

                _tree.lock_read()

                self.addCleanup(_tree.unlock)

            return trees

        finally:

            tree.unlock()

    def test_last_modified_revision_after_commit_root_unchanged(self):

        # commiting without changing the root does not change the

        # last modified except on non-rich-root-repositories.

        tree = self.make_branch_and_tree('.')

        rev1 = tree.commit('')

        rev2 = tree.commit('')

        tree1, tree2 = self._get_revtrees(tree, [rev1, rev2])

        self.assertEqual(rev1, tree1.get_file_revision(tree1.get_root_id()))

        if tree.branch.repository.supports_rich_root():

            self.assertEqual(rev1,

                tree2.get_file_revision(tree2.get_root_id()))

        else:

            self.assertEqual(rev2,

                tree2.get_file_revision(tree2.get_root_id()))

    def _add_commit_check_unchanged(self, tree, name, mini_commit=None):

        tree.add([name], [name + 'id'])

        self._commit_check_unchanged(tree, name, name + 'id',

            mini_commit=mini_commit)

    def _commit_check_unchanged(self, tree, name, file_id, mini_commit=None):

        rev1 = tree.commit('')

        if mini_commit is None:

            mini_commit = self.mini_commit

        rev2 = mini_commit(tree, name, name, False, False)

        tree1, tree2 = self._get_revtrees(tree, [rev1, rev2])

        self.assertEqual(rev1, tree1.get_file_revision(file_id))

        self.assertEqual(rev1, tree2.get_file_revision(file_id))

        expected_graph = {}

        expected_graph[(file_id, rev1)] = ()

        self.assertFileGraph(expected_graph, tree, (file_id, rev1))

    def test_last_modified_revision_after_commit_dir_unchanged(self):

        # committing without changing a dir does not change the last modified.

        tree = self.make_branch_and_tree('.')

        self.build_tree(['dir/'])

        self._add_commit_check_unchanged(tree, 'dir')

    def test_last_modified_revision_after_commit_dir_unchanged_ric(self):

        # committing without changing a dir does not change the last modified.

        tree = self.make_branch_and_tree('.')

        self.build_tree(['dir/'])

        self._add_commit_check_unchanged(tree, 'dir',

            mini_commit=self.mini_commit_record_iter_changes)

    def test_last_modified_revision_after_commit_dir_contents_unchanged(self):

        # committing without changing a dir does not change the last modified

        # of the dir even the dirs contents are changed.

        tree = self.make_branch_and_tree('.')

        self.build_tree(['dir/'])

        tree.add(['dir'], ['dirid'])

        rev1 = tree.commit('')