info.py

Python的一个ORM,现在很火

                       (2, self.obj_info, self.variable2, 10, None, False)])

        del changes1[:]
        del changes2[:]

        del self.obj.prop1
        del self.obj.prop2

        self.assertEquals(changes1,
                      [(1, self.obj_info, self.variable1, None, Undef, False),
                       (1, self.obj_info, self.variable2, None, Undef, False)])
        self.assertEquals(changes2,
                      [(2, self.obj_info, self.variable1, None, Undef, False),
                       (2, self.obj_info, self.variable2, None, Undef, False)])

    def test_add_change_notification_with_arg(self):
        changes1 = []
        changes2 = []
        def object_changed1(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes1.append((1, obj_info, variable,
                             old_value, new_value, fromdb, arg))
        def object_changed2(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes2.append((2, obj_info, variable,
                             old_value, new_value, fromdb, arg))

        self.obj_info.checkpoint()

        obj = object()
        
        self.obj_info.event.hook("changed", object_changed1, obj)
        self.obj_info.event.hook("changed", object_changed2, obj)

        self.obj.prop2 = 10
        self.obj.prop1 = 20

        self.assertEquals(changes1,
                  [(1, self.obj_info, self.variable2, Undef, 10, False, obj),
                   (1, self.obj_info, self.variable1, Undef, 20, False, obj)])
        self.assertEquals(changes2,                                        
                  [(2, self.obj_info, self.variable2, Undef, 10, False, obj),
                   (2, self.obj_info, self.variable1, Undef, 20, False, obj)])

        del changes1[:]
        del changes2[:]

        self.obj.prop1 = None
        self.obj.prop2 = None

        self.assertEquals(changes1,
                  [(1, self.obj_info, self.variable1, 20, None, False, obj),
                   (1, self.obj_info, self.variable2, 10, None, False, obj)])
        self.assertEquals(changes2,                                       
                  [(2, self.obj_info, self.variable1, 20, None, False, obj),
                   (2, self.obj_info, self.variable2, 10, None, False, obj)])

        del changes1[:]
        del changes2[:]

        del self.obj.prop1
        del self.obj.prop2

        self.assertEquals(changes1,
              [(1, self.obj_info, self.variable1, None, Undef, False, obj),
               (1, self.obj_info, self.variable2, None, Undef, False, obj)])
        self.assertEquals(changes2,                                      
              [(2, self.obj_info, self.variable1, None, Undef, False, obj),
               (2, self.obj_info, self.variable2, None, Undef, False, obj)])

    def test_remove_change_notification(self):
        changes1 = []
        changes2 = []
        def object_changed1(obj_info, variable, old_value, new_value, fromdb):
            changes1.append((1, obj_info, variable,
                             old_value, new_value, fromdb))
        def object_changed2(obj_info, variable, old_value, new_value, fromdb):
            changes2.append((2, obj_info, variable,
                             old_value, new_value, fromdb))

        self.obj_info.checkpoint()

        self.obj_info.event.hook("changed", object_changed1)
        self.obj_info.event.hook("changed", object_changed2)
        self.obj_info.event.unhook("changed", object_changed1)

        self.obj.prop2 = 20
        self.obj.prop1 = 10

        self.assertEquals(changes1, [])
        self.assertEquals(changes2,
                      [(2, self.obj_info, self.variable2, Undef, 20, False),
                       (2, self.obj_info, self.variable1, Undef, 10, False)])

    def test_remove_change_notification_with_arg(self):
        changes1 = []
        changes2 = []
        def object_changed1(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes1.append((1, obj_info, variable,
                             old_value, new_value, fromdb, arg))
        def object_changed2(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes2.append((2, obj_info, variable,
                             old_value, new_value, fromdb, arg))

        self.obj_info.checkpoint()

        obj = object()

        self.obj_info.event.hook("changed", object_changed1, obj)
        self.obj_info.event.hook("changed", object_changed2, obj)
        self.obj_info.event.unhook("changed", object_changed1, obj)

        self.obj.prop2 = 20
        self.obj.prop1 = 10

        self.assertEquals(changes1, [])
        self.assertEquals(changes2,
                  [(2, self.obj_info, self.variable2, Undef, 20, False, obj),
                   (2, self.obj_info, self.variable1, Undef, 10, False, obj)])

    def test_auto_remove_change_notification(self):
        changes1 = []
        changes2 = []
        def object_changed1(obj_info, variable, old_value, new_value, fromdb):
            changes1.append((1, obj_info, variable,
                             old_value, new_value, fromdb))
            return False
        def object_changed2(obj_info, variable, old_value, new_value, fromdb):
            changes2.append((2, obj_info, variable,
                             old_value, new_value, fromdb))
            return False

        self.obj_info.checkpoint()

        self.obj_info.event.hook("changed", object_changed1)
        self.obj_info.event.hook("changed", object_changed2)

        self.obj.prop2 = 20
        self.obj.prop1 = 10

        self.assertEquals(changes1,
                      [(1, self.obj_info, self.variable2, Undef, 20, False)])
        self.assertEquals(changes2,
                      [(2, self.obj_info, self.variable2, Undef, 20, False)])

    def test_auto_remove_change_notification_with_arg(self):
        changes1 = []
        changes2 = []
        def object_changed1(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes1.append((1, obj_info, variable,
                             old_value, new_value, fromdb, arg))
            return False
        def object_changed2(obj_info, variable,
                            old_value, new_value, fromdb, arg):
            changes2.append((2, obj_info, variable,
                             old_value, new_value, fromdb, arg))
            return False

        self.obj_info.checkpoint()

        obj = object()

        self.obj_info.event.hook("changed", object_changed1, obj)
        self.obj_info.event.hook("changed", object_changed2, obj)

        self.obj.prop2 = 20
        self.obj.prop1 = 10

        self.assertEquals(changes1,
                  [(1, self.obj_info, self.variable2, Undef, 20, False, obj)])
        self.assertEquals(changes2,
                  [(2, self.obj_info, self.variable2, Undef, 20, False, obj)])

    def test_get_obj(self):
        self.assertTrue(self.obj_info.get_obj() is self.obj)

    def test_set_obj(self):
        obj = self.Class()
        self.obj_info.set_obj(obj)
        self.assertTrue(self.obj_info.get_obj() is obj)

    def test_weak_reference(self):
        obj = self.Class()
        obj_info = get_obj_info(obj)
        del obj
        self.assertEquals(obj_info.get_obj(), None)

    def test_object_deleted_notification(self):
        obj = self.Class()
        obj_info = get_obj_info(obj)
        obj_info.tainted = True
        deleted = []
        def object_deleted(obj_info):
            deleted.append(obj_info)
        obj_info.event.hook("object-deleted", object_deleted)
        del obj_info
        del obj
        self.assertEquals(len(deleted), 1)
        self.assertEquals(getattr(deleted[0], "tainted", False), True)

    def test_object_deleted_notification_after_set_obj(self):
        obj = self.Class()
        obj_info = get_obj_info(obj)
        obj_info.tainted = True
        obj = self.Class()
        obj_info.set_obj(obj)
        deleted = []
        def object_deleted(obj_info):
            deleted.append(obj_info)
        obj_info.event.hook("object-deleted", object_deleted)
        del obj_info
        del obj
        self.assertEquals(len(deleted), 1)
        self.assertEquals(getattr(deleted[0], "tainted", False), True)


class ClassAliasTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=True)
        self.Class = Class
        self.obj = Class()
        self.ClassAlias = ClassAlias(self.Class, "alias")
        
    def test_cls_info_cls(self):
        cls_info = get_cls_info(self.ClassAlias)
        self.assertEquals(cls_info.cls, self.Class)
        self.assertEquals(cls_info.table, "alias")
        self.assertEquals(self.ClassAlias.prop1.name, "column1")
        self.assertEquals(self.ClassAlias.prop1.table, self.ClassAlias)

    def test_compile(self):
        statement, parameters = compile(self.ClassAlias)
        self.assertEquals(statement, "alias")

    def test_compile_in_select(self):
        expr = Select(self.ClassAlias.prop1, self.ClassAlias.prop1 == 1,
                      self.ClassAlias)
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT alias.column1 FROM table AS alias "
                          "WHERE alias.column1 = ?")


class TypeCompilerTest(TestHelper):

    def test_nested_classes(self):
        """Convoluted case checking that the model is right."""
        class Class1(object):
            __storm_table__ = "class1"
            id = Property(primary=True)
        class Class2(object):
            __storm_table__ = Class1
            id = Property(primary=True)
        statement, parameters = compile(Class2)
        self.assertEquals(statement, "class1")
        alias = ClassAlias(Class2, "alias")
        statement, parameters = compile(Select(alias.id))
        self.assertEquals(statement, "SELECT alias.id FROM class1 AS alias")