~bzr-pqm/bzr/bzr.dev

import changeset
from changeset import Inventory, apply_changeset, invert_dict
import os.path
from osutils import backup_file
from merge3 import Merge3

class ApplyMerge3:
    """Contents-change wrapper around merge3.Merge3"""
    def __init__(self, file_id, base, other):
        self.file_id = file_id
        self.base = base
        self.other = other
 
    def __eq__(self, other):
        if not isinstance(other, ApplyMerge3):
            return False
        return (self.base == other.base and 
                self.other == other.other and self.file_id == other.file_id)

    def __ne__(self, other):
        return not (self == other)


    def apply(self, filename, conflict_handler, reverse=False):
        new_file = filename+".new" 
        if not reverse:
            base = self.base
            other = self.other
        else:
            base = self.other
            other = self.base
        def get_lines(tree):
            if self.file_id not in tree:
                raise Exception("%s not in tree" % self.file_id)
                return ()
            return tree.get_file(self.file_id).readlines()
        base_lines = get_lines(base)
        other_lines = get_lines(other)
        m3 = Merge3(base_lines, file(filename, "rb").readlines(), other_lines)

        new_conflicts = False
        output_file = file(new_file, "wb")
        start_marker = "!START OF MERGE CONFLICT!" + "I HOPE THIS IS UNIQUE"
        for line in m3.merge_lines(name_a = "TREE", name_b = "MERGE-SOURCE", 
                       start_marker=start_marker):
            if line.startswith(start_marker):
                new_conflicts = True
                output_file.write(line.replace(start_marker, '<<<<<<<<'))
            else:
                output_file.write(line)
        output_file.close()
        if not new_conflicts:
            os.chmod(new_file, os.stat(filename).st_mode)
            os.rename(new_file, filename)
            return
        else:
            conflict_handler.merge_conflict(new_file, filename, base_lines,
                                            other_lines)


class BackupBeforeChange:
    """Contents-change wrapper to back up file first"""
    def __init__(self, contents_change):
        self.contents_change = contents_change
 
    def __eq__(self, other):
        if not isinstance(other, BackupBeforeChange):
            return False
        return (self.contents_change == other.contents_change)

    def __ne__(self, other):
        return not (self == other)

    def apply(self, filename, conflict_handler, reverse=False):
        backup_file(filename)
        self.contents_change.apply(filename, conflict_handler, reverse)


def invert_invent(inventory):
    invert_invent = {}
    for file_id in inventory:
        path = inventory.id2path(file_id)
        if path == '':
            path = './.'
        else:
            path = './' + path
        invert_invent[file_id] = path
    return invert_invent


def merge_flex(this, base, other, changeset_function, inventory_function,
               conflict_handler, merge_factory, interesting_ids):
    cset = changeset_function(base, other, interesting_ids)
    new_cset = make_merge_changeset(cset, this, base, other, 
                                    conflict_handler, merge_factory)
    result = apply_changeset(new_cset, invert_invent(this.tree.inventory),
                             this.root, conflict_handler, False)
    conflict_handler.finalize()
    return result

    

def make_merge_changeset(cset, this, base, other, 
                         conflict_handler, merge_factory):
    new_cset = changeset.Changeset()
    def get_this_contents(id):
        path = this.readonly_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, this, base, other, 
                                          conflict_handler)
            new_contents = make_merged_contents(entry, this, base, other, 
                                                conflict_handler,
                                                merge_factory)
            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

class ThreeWayConflict(Exception):
    def __init__(self, this, base, other):
        self.this = this
        self.base = base
        self.other = other
        msg = "Conflict merging %s %s and %s" % (this, base, other)
        Exception.__init__(self, msg)

def threeway_select(this, base, other):
    """Returns a value selected by the three-way algorithm.
    Raises ThreewayConflict if the algorithm yields a conflict"""
    if base == other:
        return this
    elif base == this:
        return other
    elif other == this:
        return this
    else:
        raise ThreeWayConflict(this, base, other)


def make_merged_entry(entry, this, base, other, conflict_handler):
    from bzrlib.trace import mutter
    def entry_data(file_id, tree):
        assert hasattr(tree, "__contains__"), "%s" % tree
        if not tree.has_or_had_id(file_id):
            return (None, None, "")
        entry = tree.tree.inventory[file_id]
        my_dir = tree.id2path(entry.parent_id)
        if my_dir is None:
            my_dir = ""
        return entry.name, entry.parent_id, my_dir 
    this_name, this_parent, this_dir = entry_data(entry.id, this)
    base_name, base_parent, base_dir = entry_data(entry.id, base)
    other_name, other_parent, other_dir = entry_data(entry.id, other)
    mutter("Dirs: this, base, other %r %r %r" % (this_dir, base_dir, other_dir))
    mutter("Names: this, base, other %r %r %r" % (this_name, base_name, other_name))
    old_name = this_name
    try:
        new_name = threeway_select(this_name, base_name, other_name)
    except ThreeWayConflict:
        new_name = conflict_handler.rename_conflict(entry.id, this_name, 
                                                    base_name, other_name)

    old_parent = this_parent
    try:
        new_parent = threeway_select(this_parent, base_parent, other_parent)
    except ThreeWayConflict:
        new_parent = conflict_handler.move_conflict(entry.id, this_dir,
                                                    base_dir, other_dir)
    def get_path(name, parent):
        if name is not None:
            if name == "":
                assert parent is None
                return './.'
            parent_dir = {this_parent: this_dir, other_parent: other_dir, 
                          base_parent: base_dir}
            directory = parent_dir[parent]
            return os.path.join(directory, name)
        else:
            assert parent is None
            return None

    old_path = get_path(old_name, old_parent)
        
    new_entry = changeset.ChangesetEntry(entry.id, old_parent, old_path)
    new_entry.new_path = get_path(new_name, new_parent)
    new_entry.new_parent = new_parent
    mutter(repr(new_entry))
    return new_entry


def get_contents(entry, tree):
    """Get a contents change element suitable for use with ReplaceContents
    """
    tree_entry = tree.tree.inventory[entry.id]
    if tree_entry.kind == "file":
        return changeset.FileCreate(tree.get_file(entry.id).read())
    else:
        assert tree_entry.kind in ("root_directory", "directory")
        return changeset.dir_create


def make_merged_contents(entry, this, base, other, conflict_handler,
                         merge_factory):
    contents = entry.contents_change
    if contents is None:
        return None
    this_path = this.readonly_path(entry.id)
    def make_merge():
        if this_path is None:
            return conflict_handler.missing_for_merge(entry.id, 
                                                      other.id2path(entry.id))
        return merge_factory(entry.id, base, other)

    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 = get_contents(entry, this)
                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_merge()
        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)
    

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
from bzrlib.inventory import InventoryEntry, RootEntry
from osutils import file_kind
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.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.iteritems():
            if path == dir:
                break
        if path != dir:
            raise Exception("Can't find parent for %s" % name)
        return InventoryEntry(file_id, name, kind, parent_id)


class MergeTree(object):
    def __init__(self, dir):
        self.dir = dir;
        os.mkdir(dir)
        self.inventory = {'0': ""}
        self.tree = FalseTree(self)
    
    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 remove_file(self, id):
        os.unlink(self.full_path(id))
        del self.inventory[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[id] = path

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

    def full_path(self, id):
        try:
            tree_path = self.inventory[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

    def has_or_had_id(self, file_id):
        return file_id in self

    def get_file(self, file_id):
        path = self.readonly_path(file_id)
        return file(path, "rb")

    def id2path(self, file_id):
        return self.inventory[file_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 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.FileCreate(tree.get_file(id).read())
                    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(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, merge_factory):
        conflict_handler = changeset.ExceptionConflictHandler(self.this.dir)
        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,
                                                     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(ApplyMerge3)
        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, 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")
        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(ApplyMerge3)
        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, 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)+"~")
        assert backup_exists("1")
        assert backup_exists("2")
        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)
        assert not builder.cset.entries["4"].is_boring()
        builder.change_contents("3", this="text6")
        cset = builder.merge_changeset(merge_factory)
        assert(cset.entries["1"].contents_change is not None)
        if isclass(merge_factory):
            assert(isinstance(cset.entries["1"].contents_change,
                          merge_factory))
            assert(isinstance(cset.entries["2"].contents_change,
                          merge_factory))
        assert(cset.entries["3"].is_boring())
        assert(cset.entries["4"].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)
        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()

        builder = MergeBuilder()
        builder.add_file("1", "0", "name1", "text1", 0755)
        builder.change_contents("1", other="text4", base="text3")
        builder.remove_file("1", base=True)
        self.assertRaises(changeset.NewContentsConflict,
                          builder.merge_changeset, merge_factory)
        builder.cleanup()

        builder = MergeBuilder()
        builder.add_file("1", "0", "name1", "text1", 0755)
        builder.change_contents("1", other="text4", base="text3")
        builder.remove_file("1", this=True)
        self.assertRaises(changeset.MissingForMerge, builder.merge_changeset, 
                          merge_factory)
        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(ApplyMerge3)
        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(ApplyMerge3)
        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()