/brz/remove-bazaar

# Copyright (C) 2009, 2010, 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

"""Implementation tests for breezy.merge.Merger."""

import os

from ..bzr.conflicts import TextConflict

from .. import (

    errors,

    merge as _mod_merge,

    )

from . import (

    multiply_tests,

    TestCaseWithTransport,

    )

from .test_merge_core import MergeBuilder

def load_tests(loader, standard_tests, pattern):

    """Multiply tests for tranport implementations."""

    result = loader.suiteClass()

    scenarios = [

        (name, {'merge_type': merger})

        for name, merger in _mod_merge.merge_type_registry.items()]

    return multiply_tests(standard_tests, scenarios, result)

class TestMergeImplementation(TestCaseWithTransport):

    def do_merge(self, target_tree, source_tree, **kwargs):

        merger = _mod_merge.Merger.from_revision_ids(

            target_tree, source_tree.last_revision(),

            other_branch=source_tree.branch)

        merger.merge_type = self.merge_type

        for name, value in kwargs.items():

            setattr(merger, name, value)

        merger.do_merge()

    def test_merge_specific_file(self):

        this_tree = self.make_branch_and_tree('this')

        this_tree.lock_write()

        self.addCleanup(this_tree.unlock)

        self.build_tree_contents([

            ('this/file1', b'a\nb\n'),

            ('this/file2', b'a\nb\n')

        ])

        this_tree.add(['file1', 'file2'])

        this_tree.commit('Added files')

        other_tree = this_tree.controldir.sprout('other').open_workingtree()

        self.build_tree_contents([

            ('other/file1', b'a\nb\nc\n'),

            ('other/file2', b'a\nb\nc\n')

        ])

        other_tree.commit('modified both')

        self.build_tree_contents([

            ('this/file1', b'd\na\nb\n'),

            ('this/file2', b'd\na\nb\n')

        ])

        this_tree.commit('modified both')

        self.do_merge(this_tree, other_tree, interesting_files=['file1'])

        self.assertFileEqual(b'd\na\nb\nc\n', 'this/file1')

        self.assertFileEqual(b'd\na\nb\n', 'this/file2')

    def test_merge_move_and_change(self):

        this_tree = self.make_branch_and_tree('this')

        this_tree.lock_write()

        self.addCleanup(this_tree.unlock)

        self.build_tree_contents([

            ('this/file1', b'line 1\nline 2\nline 3\nline 4\n'),

        ])

        this_tree.add('file1',)

        this_tree.commit('Added file')

        other_tree = this_tree.controldir.sprout('other').open_workingtree()

        self.build_tree_contents([

            ('other/file1', b'line 1\nline 2 to 2.1\nline 3\nline 4\n'),

        ])

        other_tree.commit('Changed 2 to 2.1')

        self.build_tree_contents([

            ('this/file1', b'line 1\nline 3\nline 2\nline 4\n'),

        ])

        this_tree.commit('Swapped 2 & 3')

        self.do_merge(this_tree, other_tree)

        if self.merge_type is _mod_merge.LCAMerger:

            self.expectFailure(

                "lca merge doesn't conflict for move and change",

                self.assertFileEqual,

                'line 1\n'

                '<<<<<<< TREE\n'

                'line 3\n'

                'line 2\n'

                '=======\n'

                'line 2 to 2.1\n'

                'line 3\n'

                '>>>>>>> MERGE-SOURCE\n'

                'line 4\n', 'this/file1')

        else:

            self.assertFileEqual(

                'line 1\n'

                '<<<<<<< TREE\n'

                'line 3\n'

                'line 2\n'

                '=======\n'

                'line 2 to 2.1\n'

                'line 3\n'

                '>>>>>>> MERGE-SOURCE\n'

                'line 4\n', 'this/file1')

    def test_modify_conflicts_with_delete(self):

        # If one side deletes a line, and the other modifies that line, then

        # the modification should be considered a conflict

        builder = self.make_branch_builder('test')

        builder.start_series()

        builder.build_snapshot(None,

                               [('add', ('', None, 'directory', None)),

                                ('add', ('foo', b'foo-id', 'file', b'a\nb\nc\nd\ne\n')),

                                ], revision_id=b'BASE-id')

        # Delete 'b\n'

        builder.build_snapshot([b'BASE-id'],

                               [('modify', ('foo', b'a\nc\nd\ne\n'))],

                               revision_id=b'OTHER-id')

        # Modify 'b\n', add 'X\n'

        builder.build_snapshot([b'BASE-id'],

                               [('modify', ('foo', b'a\nb2\nc\nd\nX\ne\n'))],

                               revision_id=b'THIS-id')

        builder.finish_series()

        branch = builder.get_branch()

        this_tree = branch.controldir.create_workingtree()

        this_tree.lock_write()

        self.addCleanup(this_tree.unlock)

        other_tree = this_tree.controldir.sprout(

            'other', b'OTHER-id').open_workingtree()

        self.do_merge(this_tree, other_tree)

        if self.merge_type is _mod_merge.LCAMerger:

            self.expectFailure("lca merge doesn't track deleted lines",

                               self.assertFileEqual,

                               'a\n'

                               '<<<<<<< TREE\n'

                               'b2\n'

                               '=======\n'

                               '>>>>>>> MERGE-SOURCE\n'

                               'c\n'

                               'd\n'

                               'X\n'

                               'e\n', 'test/foo')

        else:

            self.assertFileEqual(

                b'a\n'

                b'<<<<<<< TREE\n'

                b'b2\n'

                b'=======\n'

                b'>>>>>>> MERGE-SOURCE\n'

                b'c\n'

                b'd\n'

                b'X\n'

                b'e\n', 'test/foo')

    def get_limbodir_deletiondir(self, wt):

        transform = wt.transform()

        limbodir = transform._limbodir

        deletiondir = transform._deletiondir

        transform.finalize()

        return (limbodir, deletiondir)

    def test_merge_with_existing_limbo_empty(self):

        """Empty limbo dir is just cleaned up - see bug 427773"""

        wt = self.make_branch_and_tree('this')

        (limbodir, deletiondir) = self.get_limbodir_deletiondir(wt)

        os.mkdir(limbodir)

        self.do_merge(wt, wt)

    def test_merge_with_existing_limbo_non_empty(self):

        wt = self.make_branch_and_tree('this')

        (limbodir, deletiondir) = self.get_limbodir_deletiondir(wt)

        os.mkdir(limbodir)

        os.mkdir(os.path.join(limbodir, 'something'))

        self.assertRaises(errors.ExistingLimbo, self.do_merge, wt, wt)

        self.assertRaises(errors.LockError, wt.unlock)

    def test_merge_with_pending_deletion_empty(self):

        wt = self.make_branch_and_tree('this')

        (limbodir, deletiondir) = self.get_limbodir_deletiondir(wt)

        os.mkdir(deletiondir)

        self.do_merge(wt, wt)

    def test_merge_with_pending_deletion_non_empty(self):

        """Also see bug 427773"""

        wt = self.make_branch_and_tree('this')

        (limbodir, deletiondir) = self.get_limbodir_deletiondir(wt)

        os.mkdir(deletiondir)

        os.mkdir(os.path.join(deletiondir, 'something'))

        self.assertRaises(errors.ExistingPendingDeletion,

                          self.do_merge, wt, wt)

        self.assertRaises(errors.LockError, wt.unlock)

class TestHookMergeFileContent(TestCaseWithTransport):

    """Tests that the 'merge_file_content' hook is invoked."""

    def setUp(self):

        super(TestHookMergeFileContent, self).setUp()

        self.hook_log = []

    def install_hook_inactive(self):

        def inactive_factory(merger):

            # This hook is never active

            self.hook_log.append(('inactive',))

            return None

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', inactive_factory, 'test hook (inactive)')

    def install_hook_noop(self):

        test = self

        class HookNA(_mod_merge.AbstractPerFileMerger):

            def merge_contents(self, merge_params):

                # This hook unconditionally does nothing.

                test.hook_log.append(('no-op',))

                return 'not_applicable', None

        def hook_na_factory(merger):

            return HookNA(merger)

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', hook_na_factory, 'test hook (no-op)')

    def install_hook_success(self):

        test = self

        class HookSuccess(_mod_merge.AbstractPerFileMerger):

            def merge_contents(self, merge_params):

                test.hook_log.append(('success',))

                if merge_params.this_path == 'name1':

                    return 'success', [b'text-merged-by-hook']

                return 'not_applicable', None

        def hook_success_factory(merger):

            return HookSuccess(merger)

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', hook_success_factory, 'test hook (success)')

    def install_hook_conflict(self):

        test = self

        class HookConflict(_mod_merge.AbstractPerFileMerger):

            def merge_contents(self, merge_params):

                test.hook_log.append(('conflict',))

                if merge_params.this_path == 'name1':

                    return ('conflicted',

                            [b'text-with-conflict-markers-from-hook'])

                return 'not_applicable', None

        def hook_conflict_factory(merger):

            return HookConflict(merger)

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', hook_conflict_factory, 'test hook (delete)')

    def install_hook_delete(self):

        test = self

        class HookDelete(_mod_merge.AbstractPerFileMerger):

            def merge_contents(self, merge_params):

                test.hook_log.append(('delete',))

                if merge_params.this_path == 'name1':

                    return 'delete', None

                return 'not_applicable', None

        def hook_delete_factory(merger):

            return HookDelete(merger)

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', hook_delete_factory, 'test hook (delete)')

    def install_hook_log_lines(self):

        """Install a hook that saves the get_lines for the this, base and other

        versions of the file.

        """

        test = self

        class HookLogLines(_mod_merge.AbstractPerFileMerger):

            def merge_contents(self, merge_params):

                test.hook_log.append((

                    'log_lines',

                    merge_params.this_lines,

                    merge_params.other_lines,

                    merge_params.base_lines,

                    ))

                return 'not_applicable', None

        def hook_log_lines_factory(merger):

            return HookLogLines(merger)

        _mod_merge.Merger.hooks.install_named_hook(

            'merge_file_content', hook_log_lines_factory,

            'test hook (log_lines)')

    def make_merge_builder(self):

        builder = MergeBuilder(self.test_base_dir)

        self.addCleanup(builder.cleanup)

        return builder

    def create_file_needing_contents_merge(self, builder, name):

        file_id = name.encode('ascii') + b'-id'

        builder.add_file(file_id, builder.tree_root, name, b"text1", True)

        builder.change_contents(file_id, other=b"text4", this=b"text3")

    def test_change_vs_change(self):

        """Hook is used for (changed, changed)"""

        self.install_hook_success()

        builder = self.make_merge_builder()

        builder.add_file(b"1", builder.tree_root, "name1", b"text1", True)

        builder.change_contents(b"1", other=b"text4", this=b"text3")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [])

        with builder.this.get_file('name1') as f:

            self.assertEqual(f.read(), b'text-merged-by-hook')

    def test_change_vs_deleted(self):

        """Hook is used for (changed, deleted)"""

        self.install_hook_success()

        builder = self.make_merge_builder()

        builder.add_file(b"1", builder.tree_root, "name1", b"text1", True)

        builder.change_contents(b"1", this=b"text2")

        builder.remove_file(b"1", other=True)

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [])

        with builder.this.get_file('name1') as f:

            self.assertEqual(f.read(), b'text-merged-by-hook')

    def test_result_can_be_delete(self):

        """A hook's result can be the deletion of a file."""

        self.install_hook_delete()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [])

        self.assertFalse(builder.this.is_versioned('name1'))

        self.assertEqual([], list(builder.this.list_files()))

    def test_result_can_be_conflict(self):

        """A hook's result can be a conflict."""

        self.install_hook_conflict()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [TextConflict('name1', file_id=b'name1-id')])

        # The hook still gets to set the file contents in this case, so that it

        # can insert custom conflict markers.

        with builder.this.get_file('name1') as f:

            self.assertEqual(f.read(), b'text-with-conflict-markers-from-hook')

    def test_can_access_this_other_and_base_versions(self):

        """The hook function can call params.merger.get_lines to access the

        THIS/OTHER/BASE versions of the file.

        """

        self.install_hook_log_lines()

        builder = self.make_merge_builder()

        builder.add_file(b"1", builder.tree_root, "name1", b"text1", True)

        builder.change_contents(b"1", this=b"text2", other=b"text3")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(

            [('log_lines', [b'text2'], [b'text3'], [b'text1'])], self.hook_log)

    def test_chain_when_not_active(self):

        """When a hook function returns None, merging still works."""

        self.install_hook_inactive()

        self.install_hook_success()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [])

        with builder.this.get_file('name1') as f:

            self.assertEqual(f.read(), b'text-merged-by-hook')

        self.assertEqual([('inactive',), ('success',)], self.hook_log)

    def test_chain_when_not_applicable(self):

        """When a hook function returns not_applicable, the next function is

        tried (when one exists).

        """

        self.install_hook_noop()

        self.install_hook_success()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual(conflicts, [])

        with builder.this.get_file('name1') as f:

            self.assertEqual(f.read(), b'text-merged-by-hook')

        self.assertEqual([('no-op',), ('success',)], self.hook_log)

    def test_chain_stops_after_success(self):

        """When a hook function returns success, no later functions are tried.

        """

        self.install_hook_success()

        self.install_hook_noop()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual([('success',)], self.hook_log)

    def test_chain_stops_after_conflict(self):

        """When a hook function returns conflict, no later functions are tried.

        """

        self.install_hook_conflict()

        self.install_hook_noop()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual([('conflict',)], self.hook_log)

    def test_chain_stops_after_delete(self):

        """When a hook function returns delete, no later functions are tried.

        """

        self.install_hook_delete()

        self.install_hook_noop()

        builder = self.make_merge_builder()

        self.create_file_needing_contents_merge(builder, "name1")

        conflicts = builder.merge(self.merge_type)

        self.assertEqual([('delete',)], self.hook_log)