~bzr-pqm/bzr/bzr.dev

import changeset
from changeset import Inventory, apply_changeset, invert_dict
import os.path

class ThreewayInventory(object):
    def __init__(self, this_inventory, base_inventory, other_inventory):
        self.this = this_inventory
        self.base = base_inventory
        self.other = other_inventory
def invert_invent(inventory):
    invert_invent = {}
    for key, value in inventory.iteritems():
        invert_invent[value.id] = key
    return invert_invent

def make_inv(inventory):
    return Inventory(invert_invent(inventory))
        

def merge_flex(this, base, other, changeset_function, inventory_function,
               conflict_handler):
    this_inventory = inventory_function(this)
    base_inventory = inventory_function(base)
    other_inventory = inventory_function(other)
    inventory = ThreewayInventory(make_inv(this_inventory),
                                  make_inv(base_inventory), 
                                  make_inv(other_inventory))
    cset = changeset_function(base, other, base_inventory, other_inventory)
    new_cset = make_merge_changeset(cset, inventory, this, base, other, 
                                    conflict_handler)
    result = apply_changeset(new_cset, invert_invent(this_inventory),
                             this.root, conflict_handler, False)
    conflict_handler.finalize()
    return result

    

def make_merge_changeset(cset, inventory, this, base, other, 
                         conflict_handler=None):
    new_cset = changeset.Changeset()
    def get_this_contents(id):
        path = os.path.join(this.root, inventory.this.get_path(id))
        if os.path.isdir(path):
            return changeset.dir_create
        else:
            return changeset.FileCreate(file(path, "rb").read())

    for entry in cset.entries.itervalues():
        if entry.is_boring():
            new_cset.add_entry(entry)
        else:
            new_entry = make_merged_entry(entry, inventory, conflict_handler)
            new_contents = make_merged_contents(entry, this, base, other,
                                                conflict_handler)
            new_entry.contents_change = new_contents
            new_entry.metadata_change = make_merged_metadata(entry, base, other)
            new_cset.add_entry(new_entry)

    return new_cset

def make_merged_entry(entry, inventory, conflict_handler):
    this_name = inventory.this.get_name(entry.id)
    this_parent = inventory.this.get_parent(entry.id)
    this_dir = inventory.this.get_dir(entry.id)
    if this_dir is None:
        this_dir = ""

    base_name = inventory.base.get_name(entry.id)
    base_parent = inventory.base.get_parent(entry.id)
    base_dir = inventory.base.get_dir(entry.id)
    if base_dir is None:
        base_dir = ""
    other_name = inventory.other.get_name(entry.id)
    other_parent = inventory.other.get_parent(entry.id)
    other_dir = inventory.base.get_dir(entry.id)
    if other_dir is None:
        other_dir = ""

    if base_name == other_name:
        old_name = this_name
        new_name = this_name
    else:
        if this_name != base_name and this_name != other_name:
            conflict_handler.rename_conflict(entry.id, this_name, base_name,
                                             other_name)
        else:
            old_name = this_name
            new_name = other_name

    if base_parent == other_parent:
        old_parent = this_parent
        new_parent = this_parent
        old_dir = this_dir
        new_dir = this_dir
    else:
        if this_parent != base_parent and this_parent != other_parent:
            conflict_handler.move_conflict(entry.id, inventory)
        else:
            old_parent = this_parent
            old_dir = this_dir
            new_parent = other_parent
            new_dir = other_dir
    if old_name is not None and old_parent is not None:
        old_path = os.path.join(old_dir, old_name)
    else:
        old_path = None
    new_entry = changeset.ChangesetEntry(entry.id, old_parent, old_name)
    if new_name is not None and new_parent is not None:
        new_entry.new_path = os.path.join(new_dir, new_name)
    else:
        new_entry.new_path = None
    new_entry.new_parent = new_parent
    return new_entry


def make_merged_contents(entry, this, base, other, conflict_handler):
    contents = entry.contents_change
    if contents is None:
        return None
    this_path = this.readonly_path(entry.id)
    def make_diff3():
        if this_path is None:
            return conflict_handler.missing_for_merge(entry.id, inventory)
        base_path = base.readonly_path(entry.id)
        other_path = other.readonly_path(entry.id)    
        return changeset.Diff3Merge(base_path, other_path)

    if isinstance(contents, changeset.PatchApply):
        return make_diff3()
    if isinstance(contents, changeset.ReplaceContents):
        if contents.old_contents is None and contents.new_contents is None:
            return None
        if contents.new_contents is None:
            if this_path is not None and os.path.exists(this_path):
                return contents
            else:
                return None
        elif contents.old_contents is None:
            if this_path is None or not os.path.exists(this_path):
                return contents
            else:
                this_contents = file(this_path, "rb").read()
                if this_contents == contents.new_contents:
                    return None
                else:
                    other_path = other.readonly_path(entry.id)    
                    conflict_handler.new_contents_conflict(this_path, 
                                                           other_path)
        elif isinstance(contents.old_contents, changeset.FileCreate) and \
            isinstance(contents.new_contents, changeset.FileCreate):
            return make_diff3()
        else:
            raise Exception("Unhandled merge scenario")

def make_merged_metadata(entry, base, other):
    if entry.metadata_change is not None:
        base_path = base.readonly_path(entry.id)
        other_path = other.readonly_path(entry.id)    
        return PermissionsMerge(base_path, other_path)
    
def get_merge_entry(entry, inventory, base, other, conflict_handler):
    if entry.contents_change is not None:
        new_entry.contents_change = changeset.Diff3Merge(base_path, other_path)
    if entry.metadata_change is not None:
        new_entry.metadata_change = PermissionsMerge(base_path, other_path)

    return new_entry

class PermissionsMerge(object):
    def __init__(self, base_path, other_path):
        self.base_path = base_path
        self.other_path = other_path

    def apply(self, filename, conflict_handler, reverse=False):
        if not reverse:
            base = self.base_path
            other = self.other_path
        else:
            base = self.other_path
            other = self.base_path
        base_stat = os.stat(base).st_mode
        other_stat = os.stat(other).st_mode
        this_stat = os.stat(filename).st_mode
        if base_stat &0777 == other_stat &0777:
            return
        elif this_stat &0777 == other_stat &0777:
            return
        elif this_stat &0777 == base_stat &0777:
            os.chmod(filename, other_stat)
        else:
            conflict_handler.permission_conflict(filename, base, other)


import unittest
import tempfile
import shutil
class MergeTree(object):
    def __init__(self, dir):
        self.dir = dir;
        os.mkdir(dir)
        self.inventory = {'0': ""}
    
    def child_path(self, parent, name):
        return os.path.join(self.inventory[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[id] = path

    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[id] = path

    def abs_path(self, path):
        return os.path.join(self.dir, path)

    def full_path(self, id):
        return self.abs_path(self.inventory[id])

    def readonly_path(self, id):
        return self.full_path(id)

    def change_path(self, id, path):
        new = os.path.join(self.dir, self.inventory[id])
        os.rename(self.abs_path(self.inventory[id]), self.abs_path(path))
        self.inventory[id] = path

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_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(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_perms = os.stat(self.base.full_path(id)).st_mode &077
            new_perms = os.stat(self.other.full_path(id)).st_mode &077
            contents = changeset.ChangeUnixPermissions(old_perms, 
                                                       new_perms)
            self.cset.entries[id].metadata_change = contents

    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_perms_tree(self, id, tree, mode):
        os.chmod(tree.full_path(id), mode)

    def merge_changeset(self):
        all_inventory = ThreewayInventory(Inventory(self.this.inventory),
                                          Inventory(self.base.inventory), 
                                          Inventory(self.other.inventory))
        conflict_handler = changeset.ExceptionConflictHandler(self.this.dir)
        return make_merge_changeset(self.cset, all_inventory, self.this,
                                    self.base, self.other, conflict_handler)

    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,
                                                     self.this.dir,
                                                     conflict_handler, reverse)
        self.this.inventory =  self.apply_inv_change(inventory_change, 
                                                     self.this.inventory)

                    
        

        
    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()
        assert(cset.entries["2"].is_boring())
        assert(cset.entries["1"].name == "name1")
        assert(cset.entries["1"].new_name == "name2")
        assert(cset.entries["3"].is_boring())
        for tree in (builder.this, builder.other, builder.base):
            assert(tree.dir != builder.dir and 
                   tree.dir.startswith(builder.dir))
            for path in tree.inventory.itervalues():
                fullpath = tree.abs_path(path)
                assert(fullpath.startswith(tree.dir))
                assert(not path.startswith(tree.dir))
                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)
        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")
        assert(Inventory(builder.other.inventory).get_parent("3") == "2")
        builder.change_parent("4", this="2")
        assert(Inventory(builder.this.inventory).get_parent("4") == "2")
        builder.change_parent("5", base="2")
        assert(Inventory(builder.base.inventory).get_parent("5") == "2")
        cset = builder.merge_changeset()
        for id in ("1", "2", "4", "5"):
            assert(cset.entries[id].is_boring())
        assert(cset.entries["3"].parent == "1")
        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)
        builder.cleanup()

    def test_contents_merge(self):
        """Test diff3 merging"""
        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.change_contents("3", this="text6")
        cset = builder.merge_changeset()
        assert(cset.entries["1"].contents_change is not None)
        assert(isinstance(cset.entries["1"].contents_change,
                          changeset.Diff3Merge))
        assert(isinstance(cset.entries["2"].contents_change,
                          changeset.Diff3Merge))
        assert(cset.entries["3"].is_boring())
        builder.apply_changeset(cset)
        assert(file(builder.this.full_path("1"), "rb").read() == "text4" )
        assert(file(builder.this.full_path("2"), "rb").read() == "text2" )
        assert(os.stat(builder.this.full_path("1")).st_mode &0777 == 0755)
        assert(os.stat(builder.this.full_path("2")).st_mode &0777 == 0655)
        assert(os.stat(builder.this.full_path("3")).st_mode &0777 == 0744)
        builder.cleanup()

        builder = MergeBuilder()
        builder.add_file("1", "0", "name1", "text1", 0755)
        builder.change_contents("1", other="text4", this="text3")
        cset = builder.merge_changeset()
        self.assertRaises(changeset.MergeConflict, builder.apply_changeset,
                          cset)
        builder.cleanup()

    def test_perms_merge(self):
        builder = MergeBuilder()
        builder.add_file("1", "0", "name1", "text1", 0755)
        builder.change_perms("1", other=0655)
        builder.add_file("2", "0", "name2", "text2", 0755)
        builder.change_perms("2", base=0655)
        builder.add_file("3", "0", "name3", "text3", 0755)
        builder.change_perms("3", this=0655)
        cset = builder.merge_changeset()
        assert(cset.entries["1"].metadata_change is not None)
        assert(isinstance(cset.entries["1"].metadata_change,
                          PermissionsMerge))
        assert(isinstance(cset.entries["2"].metadata_change,
                          PermissionsMerge))
        assert(cset.entries["3"].is_boring())
        builder.apply_changeset(cset)
        assert(os.stat(builder.this.full_path("1")).st_mode &0777 == 0655)
        assert(os.stat(builder.this.full_path("2")).st_mode &0777 == 0755)
        assert(os.stat(builder.this.full_path("3")).st_mode &0777 == 0655)
        builder.cleanup();
        builder = MergeBuilder()
        builder.add_file("1", "0", "name1", "text1", 0755)
        builder.change_perms("1", other=0655, base=0555)
        cset = builder.merge_changeset()
        self.assertRaises(changeset.MergePermissionConflict, 
                     builder.apply_changeset, cset)
        builder.cleanup()

def test():        
    changeset_suite = unittest.makeSuite(MergeTest, 'test_')
    runner = unittest.TextTestRunner()
    runner.run(changeset_suite)
        
if __name__ == "__main__":
    test()