Conflict resolution with global command ordering and rebasing

gistfile1.txt

# Conflict resolution with global command ordering and rebasing

class Obj:
    def __init__(self, id):
        self.matrix = (0, 0, 0, 0)
        self.id = id

    def __repr__(self):
        return "%s" % repr(self.id)


class Doc:
    def __init__(self):
        self.objs = []

    def get_by_id(self, id):
        return [x for x in self.objs if x.id == id][0]

    def __repr__(self):
        return "Doc(%s)" % repr(self.objs)

class Op:
    def __init__(self):
        self.order = 0
        self.affected_objects = []

    def execute(self, doc):
        pass

    def __str__(self):
        return "Op(%s, %s, %s)" % (repr(self), str(self.order), str(self.affected_objects))


class AddAt(Op):
    def __init__(self, obj, obj_above_me_id):
        self.affected_objects = [obj_above_me_id]
        self.obj = obj
        self.obj_above_me_id = obj_above_me_id

    def execute(self, doc):
        try:
            obj_above_me = doc.get_by_id(self.obj_above_me_id)
        except IndexError:
            # it was deleted
            return None
        index = doc.objs.index(obj_above_me)
        doc.objs.insert(index, self.obj)

        op = Delete(self.obj)
        return op


class AddTop(Op):
    def __init__(self, obj):
        self.obj = obj
        self.affected_objects = []

    def execute(self, doc):
        doc.objs.append(self.obj)
        op = Delete(self.obj)
        return op


class Delete(Op):
    def __init__(self, obj):
        self.affected_objects = [obj.id]
        self.obj = obj

    def execute(self, doc):
        if self.obj == doc.objs[-1]:
            op = AddTop(self.obj)
        else:
            index = doc.objs.index(self.obj)
            op = AddAt(self.obj, doc.objs[index+1].id)

        doc.objs.remove(self.obj)
        return op


class User:
    def __init__(self, id, doc):
        self.id = id
        self.doc = doc
        self.undo_stack = []
        self.redo_stack = []
        self.counter = 0

    def run(self, op):
        undo = op.execute(self.doc)
        undo.order = op.order
        if undo is not None:
            self.undo_stack.append(undo)

    def run_undo(self):
        op = self.undo_stack.pop()
        redo = op.execute(self.doc)
        if redo is not None:
            redo.order = op.order
            self.redo_stack.append(redo)
        return op

    def run_redo(self):
        op = self.redo_stack.pop()
        undo = op.execute(self.doc)
        if undo is not None:
            undo.order = op.order
            self.undo_stack.append(undo)

    def run_undo_with_transform(self, other_op):
        op = self.undo_stack.pop()
        redo = op.execute(self.doc)
        if redo is not None:
            redo.order = op.order
            redo = self.transform(redo, other_op)
            self.redo_stack.append(redo)
        return op

    def run_remote(self, op):
        count_undo = 0
        while self.undo_stack and self.undo_stack[-1].order > op.order:
            self.run_undo()
            count_undo += 1
        undo = op.execute(self.doc)
        while count_undo > 0:
            self.run_redo()
            count_undo -= 1
        self.counter = max(self.counter, op.order[0]) + 1

    def add_order(self, op):
        op.order = (self.counter, self.id)
        self.counter += 1
        return op

    def __repr__(self):
        return "%s: %s" % (self.id, self.doc)


def test_add_concurrent():
    user = User("A", Doc())
    userB = User("B", Doc())

    op1 = user.add_order(AddTop(Obj(1)))
    user.run(op1)
    userB.run_remote(op1)

    obj = user.doc.objs[0]
    op2 = user.add_order(AddAt(Obj(2), obj.id))

    obj = userB.doc.objs[0]
    op3 = userB.add_order(AddAt(Obj(3), obj.id))

    assert str(user.doc) == str(userB.doc)

    user.run(op2)
    user.run_remote(op3)
    userB.run(op3)
    userB.run_remote(op2)

    print "%s\n%s\n" % (user, userB)
    assert str(user.doc) == str(userB.doc)

def test_basic_undo():
    user = User("A", Doc())
    userB = User("B", Doc())
    for i in range(0, 10):
        op1 = user.add_order(AddTop(Obj(i)))
        op1_1 = user.add_order(AddTop(Obj(i+10)))
        op2 = userB.add_order(AddTop(Obj(i+100)))
        op3 = userB.add_order(AddTop(Obj(i+200)))
        op4 = userB.add_order(AddTop(Obj(i+300)))


        user.run(op1)
        user.run(op1_1)
        userB.run(op2)
        userB.run(op3)
        userB.run(op4)

        user.run_remote(op2)
        user.run_remote(op3)
        user.run_remote(op4)
        userB.run_remote(op1)
        userB.run_remote(op1_1)

    op = user.add_order(Delete(user.doc.objs[-1]))
    user.run(op)
    userB.run(op)

    print "%s\n%s\n" % (user, userB)
    assert str(user.doc) == str(userB.doc)

    op = user.run_undo()
    userB.run_remote(op)

    print "%s\n%s\n" % (user, userB)
    assert str(user.doc) == str(userB.doc)

    while user.undo_stack:
        op = user.run_undo()
        userB.run_remote(op)
        assert str(user.doc) == str(userB.doc)

    while userB.undo_stack:
        op = userB.run_undo()
        user.run_remote(op)
        assert str(user.doc) == str(userB.doc)

    print "%s\n%s\n" % (user, userB)


def test_delete_to_be_added():
    user = User("A", Doc())
    userB = User("B", Doc())

    op1 = user.add_order(AddTop(Obj(1)))
    user.run(op1)
    userB.run_remote(op1)

    obj = user.doc.objs[0]
    op2 = user.add_order(Delete(obj))

    obj = userB.doc.objs[0]
    op3 = userB.add_order(AddAt(Obj(3), obj.id))

    user.run(op2)
    userB.run(op3)

    user.run_remote(op3)
    userB.run_remote(op2)

    print "%s\n%s\n" % (user, userB)
    assert str(user.doc) == str(userB.doc)


def test_delete_to_be_added2():
    user = User("A", Doc())
    userB = User("B", Doc())

    op1 = user.add_order(AddTop(Obj(1)))
    user.run(op1)
    userB.run_remote(op1)

    obj = user.doc.objs[0]
    obj2 = Obj(2)
    op2 = user.add_order(AddAt(obj2, obj.id))

    user.run(op2)
    userB.run_remote(op2)

    assert str(user.doc) == str(userB.doc)

    op3 = user.add_order(Delete(obj2))

    op4 = userB.add_order(AddAt(Obj(3), obj2.id))

    user.run(op3)
    userB.run(op4)

    user.run_remote(op4)
    userB.run_remote(op3)

    print "%s\n%s\n" % (user, userB)
    assert str(user.doc) == str(userB.doc)


if __name__ == "__main__":
    test_add_concurrent()
    test_basic_undo()
    test_delete_to_be_added()
    test_delete_to_be_added2()