/brz/remove-bazaar

import os

import shutil

import tempfile

import unittest

import stat

import sys

from bzrlib.selftest import TestCaseInTempDir, TestCase

from bzrlib.branch import ScratchBranch, Branch

from bzrlib.errors import (NotBranchError, NotVersionedError,

                           WorkingTreeNotRevision, BzrCommandError)

from bzrlib.inventory import RootEntry

import bzrlib.inventory as inventory

from bzrlib.osutils import file_kind, rename, sha_file

from bzrlib import changeset

from bzrlib.merge_core import (ApplyMerge3, make_merge_changeset,

                               BackupBeforeChange, ExecFlagMerge, WeaveMerge)

from bzrlib.changeset import Inventory, apply_changeset, invert_dict, \

    get_contents, ReplaceContents, ChangeExecFlag

from bzrlib.clone import copy_branch

from bzrlib.merge import merge

from bzrlib.workingtree import WorkingTree

class FalseTree(object):

    def __init__(self, realtree):

        self._realtree = realtree

        self.inventory = self

    def __getitem__(self, file_id):

        entry = self.make_inventory_entry(file_id)

        if entry is None:

            raise KeyError(file_id)

        return entry

    def make_inventory_entry(self, file_id):

        path = self._realtree.inventory_dict.get(file_id)

        if path is None:

            return None

        if path == "":

            return RootEntry(file_id)

        dir, name = os.path.split(path)

        kind = file_kind(self._realtree.abs_path(path))

        for parent_id, path in self._realtree.inventory_dict.iteritems():

            if path == dir:

                break

        if path != dir:

            raise Exception("Can't find parent for %s" % name)

        if kind not in ('directory', 'file', 'symlink'):

            raise ValueError('unknown kind %r' % kind)

        if kind == 'directory':

            return inventory.InventoryDirectory(file_id, name, parent_id)

        elif kind == 'file':

            return inventory.InventoryFile(file_id, name, parent_id)

        else:

            return inventory.InventoryLink(file_id, name, parent_id)

class MergeTree(object):

    def __init__(self, dir):

        self.dir = dir;

        os.mkdir(dir)

        self.inventory_dict = {'0': ""}

        self.inventory = FalseTree(self)

    def child_path(self, parent, name):

        return os.path.join(self.inventory_dict[parent], name)

    def add_file(self, id, parent, name, contents, mode):

        path = self.child_path(parent, name)

        full_path = self.abs_path(path)

        assert not os.path.exists(full_path)

        file(full_path, "wb").write(contents)

        os.chmod(self.abs_path(path), mode)

        self.inventory_dict[id] = path

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

        path = self.child_path(parent, name)

        full_path = self.abs_path(path)

        assert not os.path.exists(full_path)

        os.symlink(target, full_path)

        self.inventory_dict[id] = path

    def remove_file(self, id):

        os.unlink(self.full_path(id))

        del self.inventory_dict[id]

    def add_dir(self, id, parent, name, mode):

        path = self.child_path(parent, name)

        full_path = self.abs_path(path)

        assert not os.path.exists(full_path)

        os.mkdir(self.abs_path(path))

        os.chmod(self.abs_path(path), mode)

        self.inventory_dict[id] = path

    def abs_path(self, path):

        return os.path.join(self.dir, path)

    def full_path(self, id):

        try:

            tree_path = self.inventory_dict[id]

        except KeyError:

            return None

        return self.abs_path(tree_path)

    def readonly_path(self, id):

        return self.full_path(id)

    def __contains__(self, file_id):

        return file_id in self.inventory_dict

    def has_or_had_id(self, file_id):

        return file_id in self

    def get_file(self, file_id):

        path = self.full_path(file_id)

        return file(path, "rb")

    def id2path(self, file_id):

        return self.inventory_dict[file_id]

    def id2abspath(self, id):

        return self.full_path(id)

    def change_path(self, id, path):

        old_path = os.path.join(self.dir, self.inventory_dict[id])

        rename(old_path, self.abs_path(path))

        self.inventory_dict[id] = path

    def is_executable(self, file_id):

        mode = os.lstat(self.full_path(file_id)).st_mode

        return bool(stat.S_ISREG(mode) and stat.S_IEXEC & mode)

    def kind(self, file_id):

        return file_kind(self.full_path(file_id))

    def get_symlink_target(self, file_id):

        return os.readlink(self.full_path(file_id))

    def get_file_sha1(self, file_id):

        return sha_file(file(self.full_path(file_id), "rb"))

class MergeBuilder(object):

    def __init__(self):

        self.dir = tempfile.mkdtemp(prefix="BaZing")

        self.base = MergeTree(os.path.join(self.dir, "base"))

        self.this = MergeTree(os.path.join(self.dir, "this"))

        self.other = MergeTree(os.path.join(self.dir, "other"))

        self.cset = changeset.Changeset()

        self.cset.add_entry(changeset.ChangesetEntry("0", 

                                                     changeset.NULL_ID, "./."))

    def get_cset_path(self, parent, name):

        if name is None:

            assert (parent is None)

            return None

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

    def add_file(self, id, parent, name, contents, mode):

        self.base.add_file(id, parent, name, contents, mode)

        self.this.add_file(id, parent, name, contents, mode)

        self.other.add_file(id, parent, name, contents, mode)

        path = self.get_cset_path(parent, name)

        self.cset.add_entry(changeset.ChangesetEntry(id, parent, path))

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

        self.base.add_symlink(id, parent, name, contents)

        self.this.add_symlink(id, parent, name, contents)

        self.other.add_symlink(id, parent, name, contents)

        path = self.get_cset_path(parent, name)

        self.cset.add_entry(changeset.ChangesetEntry(id, parent, path))

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

        for option, tree in ((base, self.base), (this, self.this), 

                             (other, self.other)):

            if option:

                tree.remove_file(id)

            if other or base:

                change = self.cset.entries[id].contents_change

                if change is None:

                    change = changeset.ReplaceContents(None, None)

                    self.cset.entries[id].contents_change = change

                    def create_file(tree):

                        return changeset.TreeFileCreate(tree, id)

                    if not other:

                        change.new_contents = create_file(self.other)

                    if not base:

                        change.old_contents = create_file(self.base)

                else:

                    assert isinstance(change, changeset.ReplaceContents)

                if other:

                    change.new_contents=None

                if base:

                    change.old_contents=None

                if change.old_contents is None and change.new_contents is None:

                    change = None

    def add_dir(self, id, parent, name, mode):

        path = self.get_cset_path(parent, name)

        self.base.add_dir(id, parent, name, mode)

        self.cset.add_entry(changeset.ChangesetEntry(id, parent, path))

        self.this.add_dir(id, parent, name, mode)

        self.other.add_dir(id, parent, name, mode)

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

        if base is not None:

            self.change_name_tree(id, self.base, base)

            self.cset.entries[id].name = base

        if this is not None:

            self.change_name_tree(id, self.this, this)

        if other is not None:

            self.change_name_tree(id, self.other, other)

            self.cset.entries[id].new_name = other

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

        if base is not None:

            self.change_parent_tree(id, self.base, base)

            self.cset.entries[id].parent = base

            self.cset.entries[id].dir = self.cset.entries[base].path

        if this is not None:

            self.change_parent_tree(id, self.this, this)

        if other is not None:

            self.change_parent_tree(id, self.other, other)

            self.cset.entries[id].new_parent = other

            self.cset.entries[id].new_dir = \

                self.cset.entries[other].new_path

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

        if base is not None:

            self.change_contents_tree(id, self.base, base)

        if this is not None:

            self.change_contents_tree(id, self.this, this)

        if other is not None:

            self.change_contents_tree(id, self.other, other)

        if base is not None or other is not None:

            old_contents = file(self.base.full_path(id)).read()

            new_contents = file(self.other.full_path(id)).read()

            contents = changeset.ReplaceFileContents(self.base, self.other, id)

            self.cset.entries[id].contents_change = contents

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

        if base is not None:

            self.change_target_tree(id, self.base, base)

        if this is not None:

            self.change_target_tree(id, self.this, this)

        if other is not None:

            self.change_target_tree(id, self.other, other)

        if base is not None or other is not None:

            old_contents = get_contents(self.base, id)

            new_contents = get_contents(self.other, id)

            contents = ReplaceContents(old_contents, new_contents)

            self.cset.entries[id].contents_change = contents

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

        if base is not None:

            self.change_perms_tree(id, self.base, base)

        if this is not None:

            self.change_perms_tree(id, self.this, this)

        if other is not None:

            self.change_perms_tree(id, self.other, other)

        if base is not None or other is not None:

            old_exec = self.base.is_executable(id)

            new_exec = self.other.is_executable(id)

            metadata = changeset.ChangeExecFlag(old_exec, new_exec)

            self.cset.entries[id].metadata_change = metadata

    def change_name_tree(self, id, tree, name):

        new_path = tree.child_path(self.cset.entries[id].parent, name)

        tree.change_path(id, new_path)

    def change_parent_tree(self, id, tree, parent):

        new_path = tree.child_path(parent, self.cset.entries[id].name)

        tree.change_path(id, new_path)

    def change_contents_tree(self, id, tree, contents):

        path = tree.full_path(id)

        mode = os.stat(path).st_mode

        file(path, "w").write(contents)

        os.chmod(path, mode)

    def change_target_tree(self, id, tree, target):

        path = tree.full_path(id)

        os.unlink(path)

        os.symlink(target, path)

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

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

    def merge_changeset(self, merge_factory):

        conflict_handler = changeset.ExceptionConflictHandler()

        return make_merge_changeset(self.cset, self.this, self.base,

                                    self.other, conflict_handler,

                                    merge_factory)

    def apply_inv_change(self, inventory_change, orig_inventory):

        orig_inventory_by_path = {}

        for file_id, path in orig_inventory.iteritems():

            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 inventory_change.has_key(file_id):

                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 os.path.join(dirname, orig_inventory[file_id])

        new_inventory = {}

        for file_id in orig_inventory.iterkeys():

            path = new_path(file_id)

            if path is None:

                continue

            new_inventory[file_id] = path

        for file_id, path in inventory_change.iteritems():

            if orig_inventory.has_key(file_id):

                continue

            new_inventory[file_id] = path

        return new_inventory

    def apply_changeset(self, cset, conflict_handler=None, reverse=False):

        inventory_change = changeset.apply_changeset(cset,

                                                     self.this.inventory_dict,

                                                     self.this.dir,

                                                     conflict_handler, reverse)

        self.this.inventory_dict =  self.apply_inv_change(inventory_change, 

                                                     self.this.inventory_dict)

    def cleanup(self):

        shutil.rmtree(self.dir)

class MergeTest(unittest.TestCase):

    def test_change_name(self):

        """Test renames"""

        builder = MergeBuilder()

        builder.add_file("1", "0", "name1", "hello1", 0755)

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

        builder.add_file("2", "0", "name3", "hello2", 0755)

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

        builder.add_file("3", "0", "name5", "hello3", 0755)

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

        cset = builder.merge_changeset(ApplyMerge3)

        self.assert_(cset.entries["2"].is_boring())

        self.assert_(cset.entries["1"].name == "name1")

        self.assert_(cset.entries["1"].new_name == "name2")

        self.assert_(cset.entries["3"].is_boring())

        for tree in (builder.this, builder.other, builder.base):

            self.assert_(tree.dir != builder.dir and 

                   tree.dir.startswith(builder.dir))

            for path in tree.inventory_dict.itervalues():

                fullpath = tree.abs_path(path)

                self.assert_(fullpath.startswith(tree.dir))

                self.assert_(not path.startswith(tree.dir))

                self.assert_(os.path.exists(fullpath))

        builder.apply_changeset(cset)

        builder.cleanup()

        builder = MergeBuilder()

        builder.add_file("1", "0", "name1", "hello1", 0644)

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

        self.assertRaises(changeset.RenameConflict, 

                          builder.merge_changeset, ApplyMerge3)

        builder.cleanup()

    def test_file_moves(self):

        """Test moves"""

        builder = MergeBuilder()

        builder.add_dir("1", "0", "dir1", 0755)

        builder.add_dir("2", "0", "dir2", 0755)

        builder.add_file("3", "1", "file1", "hello1", 0644)

        builder.add_file("4", "1", "file2", "hello2", 0644)

        builder.add_file("5", "1", "file3", "hello3", 0644)

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

        self.assert_(Inventory(builder.other.inventory_dict).get_parent("3") == "2")

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

        self.assert_(Inventory(builder.this.inventory_dict).get_parent("4") == "2")

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

        self.assert_(Inventory(builder.base.inventory_dict).get_parent("5") == "2")

        cset = builder.merge_changeset(ApplyMerge3)

        for id in ("1", "2", "4", "5"):

            self.assert_(cset.entries[id].is_boring())

        self.assert_(cset.entries["3"].parent == "1")

        self.assert_(cset.entries["3"].new_parent == "2")

        builder.apply_changeset(cset)

        builder.cleanup()

        builder = MergeBuilder()

        builder.add_dir("1", "0", "dir1", 0755)

        builder.add_dir("2", "0", "dir2", 0755)

        builder.add_dir("3", "0", "dir3", 0755)

        builder.add_file("4", "1", "file1", "hello1", 0644)

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

        self.assertRaises(changeset.MoveConflict, 

                          builder.merge_changeset, ApplyMerge3)

        builder.cleanup()

    def test_contents_merge(self):

        """Test merge3 merging"""

        self.do_contents_test(ApplyMerge3)

    def test_contents_merge2(self):

        """Test diff3 merging"""

        self.do_contents_test(changeset.Diff3Merge)

    def test_contents_merge3(self):

        """Test diff3 merging"""

        def backup_merge(file_id, base, other):

            return BackupBeforeChange(ApplyMerge3(file_id, base, other))

        builder = self.contents_test_success(backup_merge)

        def backup_exists(file_id):

            return os.path.exists(builder.this.full_path(file_id)+"~")

        self.assert_(backup_exists("1"))

        self.assert_(backup_exists("2"))

        self.assert_(not backup_exists("3"))

        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):

        from inspect import isclass

        builder = MergeBuilder()

        builder.add_file("1", "0", "name1", "text1", 0755)

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

        builder.add_file("2", "0", "name3", "text2", 0655)

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

        builder.add_file("3", "0", "name5", "text3", 0744)

        builder.add_file("4", "0", "name6", "text4", 0744)

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

        self.assert_(not builder.cset.entries["4"].is_boring())

        builder.change_contents("3", this="text6")

        cset = builder.merge_changeset(merge_factory)

        self.assert_(cset.entries["1"].contents_change is not None)

        if isclass(merge_factory):

            self.assert_(isinstance(cset.entries["1"].contents_change,

                          merge_factory))

            self.assert_(isinstance(cset.entries["2"].contents_change,

                          merge_factory))

        self.assert_(cset.entries["3"].is_boring())

        self.assert_(cset.entries["4"].is_boring())

        builder.apply_changeset(cset)

        self.assert_(file(builder.this.full_path("1"), "rb").read() == "text4" )

        self.assert_(file(builder.this.full_path("2"), "rb").read() == "text2" )

        if sys.platform != "win32":

            self.assert_(os.stat(builder.this.full_path("1")).st_mode &0777 == 0755)

            self.assert_(os.stat(builder.this.full_path("2")).st_mode &0777 == 0655)

            self.assert_(os.stat(builder.this.full_path("3")).st_mode &0777 == 0744)

        return builder

    def contents_test_conflicts(self, merge_factory):

        builder = MergeBuilder()

        builder.add_file("1", "0", "name1", "text1", 0755)

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

        cset = builder.merge_changeset(merge_factory)

        self.assertRaises(changeset.MergeConflict, builder.apply_changeset,

                          cset)

        builder.cleanup()

    def test_symlink_conflicts(self):

        if sys.platform != "win32":

            builder = MergeBuilder()

            builder.add_symlink("2", "0", "name2", "target1")

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

            self.assertRaises(changeset.ThreewayContentsConflict,

                              builder.merge_changeset, ApplyMerge3)

            builder.cleanup()

    def test_symlink_merge(self):

        if sys.platform != "win32":

            builder = MergeBuilder()

            builder.add_symlink("1", "0", "name1", "target1")

            builder.add_symlink("2", "0", "name2", "target1")

            builder.add_symlink("3", "0", "name3", "target1")

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

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

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

            self.assertNotEqual(builder.cset.entries['2'].contents_change,

                                builder.cset.entries['3'].contents_change)

            cset = builder.merge_changeset(ApplyMerge3)

            builder.apply_changeset(cset)

            self.assertEqual(builder.this.get_symlink_target("1"), "target2")

            self.assertEqual(builder.this.get_symlink_target("2"), "target1")

            self.assertEqual(builder.this.get_symlink_target("3"), "target2")

            builder.cleanup()

    def test_perms_merge(self):

        builder = MergeBuilder()

        builder.add_file("1", "0", "name1", "text1", 0755)

        builder.change_perms("1", other=0644)

        builder.add_file("2", "0", "name2", "text2", 0755)

        builder.change_perms("2", base=0644)

        builder.add_file("3", "0", "name3", "text3", 0755)

        builder.change_perms("3", this=0644)

        cset = builder.merge_changeset(ApplyMerge3)

        self.assert_(cset.entries["1"].metadata_change is not None)

        self.assert_(isinstance(cset.entries["1"].metadata_change, ExecFlagMerge))

        self.assert_(isinstance(cset.entries["2"].metadata_change, ExecFlagMerge))

        self.assert_(cset.entries["3"].is_boring())

        builder.apply_changeset(cset)

        if sys.platform != "win32":

            self.assert_(os.lstat(builder.this.full_path("1")).st_mode &0100 == 0000)