/brz/remove-bazaar

# Copyright (C) 2005-2012, 2016 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

import os

import sys

import breezy

from .. import (

    controldir,

    errors,

    generate_ids,

    merge_directive,

    osutils,

    )

from ..conflicts import (

    ContentsConflict,

    TextConflict,

    PathConflict,

    )

from ..merge import (

    Merge3Merger,

    Diff3Merger,

    WeaveMerger,

    Merger,

    )

from ..osutils import getcwd, pathjoin

from ..transform import TreeTransform

from . import TestCaseWithTransport, TestSkipped

from ..workingtree import WorkingTree

class MergeBuilder(object):

    def __init__(self, dir=None):

        self.dir = osutils.mkdtemp(prefix="merge-test", dir=dir)

        self.tree_root = generate_ids.gen_root_id()

        def wt(name):

            path = pathjoin(self.dir, name)

            os.mkdir(path)

            wt = controldir.ControlDir.create_standalone_workingtree(path)

            # the tests perform pulls, so need a branch that is writeable.

            wt.lock_write()

            wt.set_root_id(self.tree_root)

            wt.flush()

            tt = TreeTransform(wt)

            return wt, tt

        self.base, self.base_tt = wt('base')

        self.this, self.this_tt = wt('this')

        self.other, self.other_tt = wt('other')

    def get_cset_path(self, parent, name):

        if name is None:

            if parent is not None:

                raise AssertionError()

            return None

        return pathjoin(self.cset.entries[parent].path, name)

    def add_file(self, id, parent, name, contents, executable, this=True,

                 base=True, other=True):

        def new_file(tt):

            parent_id = tt.trans_id_file_id(parent)

            tt.new_file(name, parent_id, [contents], id, executable)

        for option, tt in self.selected_transforms(this, base, other):

            if option is True:

                new_file(tt)

    def merge(self, merge_type=Merge3Merger, interesting_files=None, **kwargs):

        merger = self.make_merger(merge_type, interesting_files, **kwargs)

        merger.do_merge()

        return merger.cooked_conflicts

    def make_preview_transform(self):

        merger = self.make_merger(Merge3Merger, None, this_revision_tree=True)

        return merger.make_preview_transform()

    def make_merger(self, merge_type, interesting_files,

                    this_revision_tree=False, **kwargs):

        self.base_tt.apply()

        self.base.commit('base commit')

        for tt, wt in ((self.this_tt, self.this), (self.other_tt, self.other)):

            # why does this not do wt.pull() ?

            wt.branch.pull(self.base.branch)

            wt.set_parent_ids([wt.branch.last_revision()])

            wt.flush()

            # We maintain a write lock, so make sure changes are flushed to

            # disk first

            tt.apply()

            wt.commit('branch commit')

            wt.flush()

            if wt.branch.last_revision_info()[0] != 2:

                raise AssertionError()

        self.this.branch.fetch(self.other.branch)

        other_basis = self.other.branch.basis_tree()

        if this_revision_tree:

            self.this.commit('message')

            this_tree = self.this.basis_tree()

        else:

            this_tree = self.this

        merger = merge_type(this_tree, self.this, self.base, other_basis,

                            interesting_files=interesting_files, do_merge=False,

                            this_branch=self.this.branch, **kwargs)

        return merger

    def list_transforms(self):

        return [self.this_tt, self.base_tt, self.other_tt]

    def selected_transforms(self, this, base, other):

        pairs = [(this, self.this_tt), (base, self.base_tt),

                 (other, self.other_tt)]

        return [(v, tt) for (v, tt) in pairs if v is not None]

    def add_symlink(self, id, parent, name, contents):

        for tt in self.list_transforms():

            parent_id = tt.trans_id_file_id(parent)

            tt.new_symlink(name, parent_id, contents, id)

    def remove_file(self, file_id, base=False, this=False, other=False):

        for option, tt in self.selected_transforms(this, base, other):

            if option is True:

                trans_id = tt.trans_id_file_id(file_id)

                tt.cancel_creation(trans_id)

                tt.cancel_versioning(trans_id)

                tt.set_executability(None, trans_id)

    def add_dir(self, file_id, parent, name, this=True, base=True, other=True):

        for option, tt in self.selected_transforms(this, base, other):

            if option is True:

                parent_id = tt.trans_id_file_id(parent)

                tt.new_directory(name, parent_id, file_id)

    def change_name(self, id, base=None, this=None, other=None):

        for val, tt in ((base, self.base_tt), (this, self.this_tt),

                        (other, self.other_tt)):

            if val is None:

                continue

            trans_id = tt.trans_id_file_id(id)

            parent_id = tt.final_parent(trans_id)

            tt.adjust_path(val, parent_id, trans_id)

    def change_parent(self, file_id, base=None, this=None, other=None):

        for parent, tt in self.selected_transforms(this, base, other):

            trans_id = tt.trans_id_file_id(file_id)

            parent_id = tt.trans_id_file_id(parent)

            tt.adjust_path(tt.final_name(trans_id), parent_id, trans_id)

    def change_contents(self, file_id, base=None, this=None, other=None):

        for contents, tt in self.selected_transforms(this, base, other):

            trans_id = tt.trans_id_file_id(file_id)

            tt.cancel_creation(trans_id)

            tt.create_file([contents], trans_id)

    def change_target(self, id, base=None, this=None, other=None):

        for target, tt in self.selected_transforms(this, base, other):

            trans_id = tt.trans_id_file_id(id)

            tt.cancel_creation(trans_id)

            tt.create_symlink(target, trans_id)

    def change_perms(self, id, base=None, this=None, other=None):

        for executability, tt in self.selected_transforms(this, base, other):

            trans_id = tt.trans_id_file_id(id)

            tt.set_executability(None, trans_id)

            tt.set_executability(executability, trans_id)

    def change_perms_tree(self, id, tree, mode):

        os.chmod(tree.full_path(id), mode)

    def apply_inv_change(self, inventory_change, orig_inventory):

        orig_inventory_by_path = {}

        for file_id, path in orig_inventory.items():

            orig_inventory_by_path[path] = file_id

        def parent_id(file_id):

            try:

                parent_dir = os.path.dirname(orig_inventory[file_id])

            except:

                print(file_id)

                raise

            if parent_dir == "":

                return None

            return orig_inventory_by_path[parent_dir]

        def new_path(file_id):

            if fild_id in inventory_change:

                return inventory_change[file_id]

            else:

                parent = parent_id(file_id)

                if parent is None:

                    return orig_inventory[file_id]

                dirname = new_path(parent)

                return pathjoin(dirname, os.path.basename(orig_inventory[file_id]))

        new_inventory = {}

        for file_id in orig_inventory:

            path = new_path(file_id)

            if path is None:

                continue

            new_inventory[file_id] = path

        for file_id, path in inventory_change.items():

            if file_id in orig_inventory:

                continue

            new_inventory[file_id] = path

        return new_inventory

    def unlock(self):

        self.base.unlock()

        self.this.unlock()

        self.other.unlock()

    def cleanup(self):

        self.unlock()

        osutils.rmtree(self.dir)

class MergeTest(TestCaseWithTransport):

    def test_change_name(self):

        """Test renames"""

        builder = MergeBuilder(getcwd())

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

        builder.change_name(b"1", other="name2")

        builder.add_file(b"2", builder.tree_root, "name3", b"hello2", True)

        builder.change_name(b"2", base="name4")

        builder.add_file(b"3", builder.tree_root, "name5", b"hello3", True)

        builder.change_name(b"3", this="name6")

        builder.merge()

        builder.cleanup()

        builder = MergeBuilder(getcwd())

        builder.add_file(b"1", builder.tree_root, "name1", b"hello1", False)

        builder.change_name(b"1", other="name2", this="name3")

        conflicts = builder.merge()

        self.assertEqual(conflicts, [PathConflict('name3', 'name2', b'1')])

        builder.cleanup()

    def test_merge_one(self):

        builder = MergeBuilder(getcwd())

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

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

        builder.add_file(b"2", builder.tree_root, "name2", b"hello1", True)

        builder.change_contents(b"2", other=b"text4")

        builder.merge(interesting_files=["name1"])

        self.assertEqual(builder.this.get_file("name1").read(), b"text4")

        self.assertEqual(builder.this.get_file("name2").read(), b"hello1")

        builder.cleanup()

    def test_file_moves(self):

        """Test moves"""

        builder = MergeBuilder(getcwd())

        builder.add_dir(b"1", builder.tree_root, "dir1")

        builder.add_dir(b"2", builder.tree_root, "dir2")

        builder.add_file(b"3", b"1", "file1", b"hello1", True)

        builder.add_file(b"4", b"1", "file2", b"hello2", True)

        builder.add_file(b"5", b"1", "file3", b"hello3", True)

        builder.change_parent(b"3", other=b"2")

        builder.change_parent(b"4", this=b"2")

        builder.change_parent(b"5", base=b"2")

        builder.merge()

        builder.cleanup()

        builder = MergeBuilder(getcwd())

        builder.add_dir(b"1", builder.tree_root, "dir1")

        builder.add_dir(b"2", builder.tree_root, "dir2")

        builder.add_dir(b"3", builder.tree_root, "dir3")

        builder.add_file(b"4", b"1", "file1", b"hello1", False)

        builder.change_parent(b"4", other=b"2", this=b"3")

        conflicts = builder.merge()

        path2 = pathjoin('dir2', 'file1')

        path3 = pathjoin('dir3', 'file1')

        self.assertEqual(conflicts, [PathConflict(path3, path2, b'4')])

        builder.cleanup()

    def test_contents_merge(self):

        """Test merge3 merging"""

        self.do_contents_test(Merge3Merger)

    def test_contents_merge2(self):

        """Test diff3 merging"""

        if sys.platform == 'win32':

            raise TestSkipped("diff3 does not have --binary flag"

                              " and therefore always fails on win32")

        try:

            self.do_contents_test(Diff3Merger)

        except errors.NoDiff3:

            raise TestSkipped("diff3 not available")

    def test_contents_merge3(self):

        """Test diff3 merging"""

        self.do_contents_test(WeaveMerger)

    def test_reprocess_weave(self):

        # Reprocess works on weaves, and behaves as expected

        builder = MergeBuilder(getcwd())

        builder.add_file(b'a', builder.tree_root, 'blah', b'a', False)

        builder.change_contents(

            b'a', this=b'b\nc\nd\ne\n', other=b'z\nc\nd\ny\n')

        builder.merge(WeaveMerger, reprocess=True)

        expected = b"""<<<<<<< TREE

b

=======

z

>>>>>>> MERGE-SOURCE

c

d

<<<<<<< TREE

e

=======

y

>>>>>>> MERGE-SOURCE

"""

        self.assertEqualDiff(

            builder.this.get_file(builder.this.id2path(b"a")).read(),

            expected)

        builder.cleanup()

    def do_contents_test(self, merge_factory):

        """Test merging with specified ContentsChange factory"""

        builder = self.contents_test_success(merge_factory)

        builder.cleanup()

        self.contents_test_conflicts(merge_factory)

    def contents_test_success(self, merge_factory):

        builder = MergeBuilder(getcwd())

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

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

        builder.add_file(b"2", builder.tree_root, "name3", b"text2", False)

        builder.change_contents(b"2", base=b"text5")

        builder.add_file(b"3", builder.tree_root, "name5", b"text3", True)

        builder.add_file(b"4", builder.tree_root, "name6", b"text4", True)

        builder.remove_file(b"4", base=True)

        builder.add_file(b"5", builder.tree_root, "name7", b"a\nb\nc\nd\ne\nf\n",

                         True)

        builder.change_contents(b"5", other=b"a\nz\nc\nd\ne\nf\n",

                                this=b"a\nb\nc\nd\ne\nz\n")

        conflicts = builder.merge(merge_factory)

        try:

            self.assertEqual([], conflicts)

            self.assertEqual(b"text4", builder.this.get_file("name1").read())

            self.assertEqual(b"text2", builder.this.get_file("name3").read())

            self.assertEqual(b"a\nz\nc\nd\ne\nz\n",

                             builder.this.get_file("name7").read())

            self.assertTrue(builder.this.is_executable("name1"))

            self.assertFalse(builder.this.is_executable("name3"))

            self.assertTrue(builder.this.is_executable("name5"))

        except:

            builder.unlock()

            raise

        return builder

    def contents_test_conflicts(self, merge_factory):

        builder = MergeBuilder(getcwd())

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

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

        builder.add_file(b"2", builder.tree_root, "name2", b"text1", True)

        builder.change_contents(b"2", other=b"\x00", this=b"text3")

        builder.add_file(b"3", builder.tree_root, "name3", b"text5", False)

        builder.change_perms(b"3", this=True)

        builder.change_contents(b'3', this=b'moretext')

        builder.remove_file(b'3', other=True)

        conflicts = builder.merge(merge_factory)

        self.assertEqual(conflicts, [TextConflict('name1', file_id=b'1'),

                                     ContentsConflict('name2', file_id=b'2'),

                                     ContentsConflict('name3', file_id=b'3')])

        with builder.this.get_file(builder.this.id2path(b'2')) as f:

            self.assertEqual(f.read(), b'\x00')

        builder.cleanup()

    def test_symlink_conflicts(self):

        if sys.platform != "win32":

            builder = MergeBuilder(getcwd())

            builder.add_symlink(b"2", builder.tree_root, "name2", "target1")

            builder.change_target(b"2", other="target4", base="text3")

            conflicts = builder.merge()

            self.assertEqual(conflicts, [ContentsConflict('name2',

                                                          file_id=b'2')])

            builder.cleanup()

    def test_symlink_merge(self):

        if sys.platform != "win32":

            builder = MergeBuilder(getcwd())

            builder.add_symlink(b"1", builder.tree_root, "name1", "target1")

            builder.add_symlink(b"2", builder.tree_root, "name2", "target1")

            builder.add_symlink(b"3", builder.tree_root, "name3", "target1")

            builder.change_target(b"1", this=b"target2")

            builder.change_target(b"2", base=b"target2")

            builder.change_target(b"3", other=b"target2")

            builder.merge()

            self.assertEqual(

                builder.this.get_symlink_target("name1"), "target2")

            self.assertEqual(

                builder.this.get_symlink_target("name2"), "target1")

            self.assertEqual(

                builder.this.get_symlink_target("name3"), "target2")

            builder.cleanup()

    def test_no_passive_add(self):

        builder = MergeBuilder(getcwd())

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

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

        builder.merge()

        builder.cleanup()

    def test_perms_merge(self):

        builder = MergeBuilder(getcwd())

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

        builder.change_perms(b"1", other=False)

        builder.add_file(b"2", builder.tree_root, "name2", b"text2", True)

        builder.change_perms(b"2", base=False)

        builder.add_file(b"3", builder.tree_root, "name3", b"text3", True)

        builder.change_perms(b"3", this=False)

        builder.add_file(b'4', builder.tree_root, 'name4', b'text4', False)

        builder.change_perms(b'4', this=True)

        builder.remove_file(b'4', base=True)

        builder.merge()

        self.assertIs(builder.this.is_executable("name1"), False)

        self.assertIs(builder.this.is_executable("name2"), True)

        self.assertIs(builder.this.is_executable("name3"), False)

        builder.cleanup()

    def test_new_suffix(self):

        builder = MergeBuilder(getcwd())

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

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

        builder.add_file(b"2", builder.tree_root, "name1.new", b"text2", True)

        builder.merge()

        os.lstat(builder.this.abspath(builder.this.id2path(b"2")))

        builder.cleanup()

    def test_spurious_conflict(self):

        builder = MergeBuilder(getcwd())

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

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

        builder.add_file(b"2", builder.tree_root, "name1", b"text1", False,

                         this=False, base=False)

        conflicts = builder.merge()

        self.assertEqual(conflicts, [])

        builder.cleanup()

    def test_merge_one_renamed(self):

        builder = MergeBuilder(getcwd())

        builder.add_file(b'1', builder.tree_root, 'name1', b'text1a', False)

        builder.change_name(b'1', this='name2')

        builder.change_contents(b'1', other=b'text2')

        builder.merge(interesting_files=['name2'])

        self.assertEqual(b'text2', builder.this.get_file('name2').read())

        builder.cleanup()

class FunctionalMergeTest(TestCaseWithTransport):

    def test_trivial_star_merge(self):

        """Test that merges in a star shape Just Work."""

        # John starts a branch

        self.build_tree(("original/", "original/file1", "original/file2"))

        tree = self.make_branch_and_tree('original')

        branch = tree.branch

        tree.smart_add(["original"])

        tree.commit("start branch.", verbose=False)

        # Mary branches it.

        self.build_tree(("mary/",))

        branch.controldir.clone("mary")

        # Now John commits a change

        with open("original/file1", "wt") as f:

            f.write("John\n")

        tree.commit("change file1")

        # Mary does too

        mary_tree = WorkingTree.open('mary')

        mary_branch = mary_tree.branch

        with open("mary/file2", "wt") as f:

            f.write("Mary\n")

        mary_tree.commit("change file2")

        # john should be able to merge with no conflicts.

        base = [None, None]

        other = ("mary", -1)

        tree.merge_from_branch(mary_tree.branch)

        with open("original/file1", "rt") as f:

            self.assertEqual("John\n", f.read())

        with open("original/file2", "rt") as f:

            self.assertEqual("Mary\n", f.read())

    def test_conflicts(self):

        wta = self.make_branch_and_tree('a')

        self.build_tree_contents([('a/file', b'contents\n')])