info.py

Python的一个ORM,现在很火

#
# Copyright (c) 2006, 2007 Canonical
#
# Written by Gustavo Niemeyer <[e-mail]>
#
# This file is part of Storm Object Relational Mapper.
#
# Storm is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation; either version 2.1 of
# the License, or (at your option) any later version.
#
# Storm is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
import gc

from storm.exceptions import ClassInfoError
from storm.properties import Property
from storm.variables import Variable
from storm.expr import Undef, Select, compile
from storm.info import *

from tests.helper import TestHelper


class GetTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=True)
        self.Class = Class
        self.obj = Class()

    def test_get_cls_info(self):
        cls_info = get_cls_info(self.Class)
        self.assertTrue(isinstance(cls_info, ClassInfo))
        self.assertTrue(cls_info is get_cls_info(self.Class))

    def test_get_obj_info(self):
        obj_info = get_obj_info(self.obj)
        self.assertTrue(isinstance(obj_info, ObjectInfo))
        self.assertTrue(obj_info is get_obj_info(self.obj))

    def test_get_obj_info_on_obj_info(self):
        obj_info = get_obj_info(self.obj)
        self.assertTrue(get_obj_info(obj_info) is obj_info)

    def test_set_obj_info(self):
        obj_info1 = get_obj_info(self.obj)
        obj_info2 = ObjectInfo(self.obj)
        self.assertEquals(get_obj_info(self.obj), obj_info1)
        set_obj_info(self.obj, obj_info2)
        self.assertEquals(get_obj_info(self.obj), obj_info2)

    def test_get_info(self):
        obj_info1, cls_info1 = get_info(self.obj)
        obj_info2, cls_info2 = get_info(self.obj)
        self.assertTrue(isinstance(obj_info1, ObjectInfo))
        self.assertTrue(isinstance(cls_info1, ClassInfo))
        self.assertTrue(obj_info1 is obj_info2)
        self.assertTrue(cls_info1 is cls_info2)

    def test_get_info_on_obj_info(self):
        obj_info1 = get_obj_info(self.obj)
        cls_info1 = get_cls_info(self.Class)
        obj_info2, cls_info2 = get_info(obj_info1)
        self.assertTrue(obj_info2 is obj_info1)
        self.assertTrue(cls_info2 is cls_info1)


class ClassInfoTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=True)
            prop2 = Property("column2")
        self.Class = Class
        self.cls_info = get_cls_info(Class)

    def test_invalid_class(self):
        class Class(object): pass
        self.assertRaises(ClassInfoError, ClassInfo, Class)

    def test_cls(self):
        self.assertEquals(self.cls_info.cls, self.Class)

    def test_columns(self):
        self.assertEquals(self.cls_info.columns,
                          (self.Class.prop1, self.Class.prop2))

    def test_table(self):
        self.assertEquals(self.cls_info.table, "table")

    def test_primary_key(self):
        # Can't use == for props.
        self.assertTrue(self.cls_info.primary_key[0] is self.Class.prop1)
        self.assertEquals(len(self.cls_info.primary_key), 1)

    def test_primary_key_with_attribute(self):
        class SubClass(self.Class):
            __storm_primary__ = "prop2"

        cls_info = get_cls_info(SubClass)

        self.assertTrue(cls_info.primary_key[0] is SubClass.prop2)
        self.assertEquals(len(self.cls_info.primary_key), 1)

    def test_primary_key_composed(self):
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=2)
            prop2 = Property("column2", primary=1)
        cls_info = ClassInfo(Class)

        # Can't use == for props, since they're columns.
        self.assertTrue(cls_info.primary_key[0] is Class.prop2)
        self.assertTrue(cls_info.primary_key[1] is Class.prop1)
        self.assertEquals(len(cls_info.primary_key), 2)

    def test_primary_key_composed_with_attribute(self):
        class Class(object):
            __storm_table__ = "table"
            __storm_primary__ = "prop2", "prop1"
            # Define primary=True to ensure that the attribute
            # has precedence.
            prop1 = Property("column1", primary=True)
            prop2 = Property("column2")
        cls_info = ClassInfo(Class)

        # Can't use == for props, since they're columns.
        self.assertTrue(cls_info.primary_key[0] is Class.prop2)
        self.assertTrue(cls_info.primary_key[1] is Class.prop1)
        self.assertEquals(len(cls_info.primary_key), 2)

    def test_primary_key_composed_duplicated(self):
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=True)
            prop2 = Property("column2", primary=True)
        self.assertRaises(ClassInfoError, ClassInfo, Class)

    def test_primary_key_missing(self):
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1")
            prop2 = Property("column2")
        self.assertRaises(ClassInfoError, ClassInfo, Class)

    def test_primary_key_attribute_missing(self):
        class Class(object):
            __storm_table__ = "table"
            __storm_primary__ = ()
            prop1 = Property("column1", primary=True)
            prop2 = Property("column2")
        self.assertRaises(ClassInfoError, ClassInfo, Class)

    def test_primary_key_pos(self):
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=2)
            prop2 = Property("column2")
            prop3 = Property("column3", primary=1)
        cls_info = ClassInfo(Class)
        self.assertEquals(cls_info.primary_key_pos, (2, 0))


class ObjectInfoTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)
        class Class(object):
            __storm_table__ = "table"
            prop1 = Property("column1", primary=True)
            prop2 = Property("column2")
        self.Class = Class
        self.obj = Class()
        self.obj_info = get_obj_info(self.obj)
        self.cls_info = get_cls_info(Class)
        self.variable1 = self.obj_info.variables[Class.prop1]
        self.variable2 = self.obj_info.variables[Class.prop2]


    def test_dict_subclass(self):
        self.assertTrue(isinstance(self.obj_info, dict))

    def test_variables(self):
        self.assertTrue(isinstance(self.obj_info.variables, dict))
        
        for column in self.cls_info.columns:
            variable = self.obj_info.variables.get(column)
            self.assertTrue(isinstance(variable, Variable))

        self.assertEquals(len(self.obj_info.variables),
                          len(self.cls_info.columns))

    def test_primary_vars(self):
        self.assertTrue(isinstance(self.obj_info.primary_vars, tuple))
        
        for column, variable in zip(self.cls_info.primary_key,
                                    self.obj_info.primary_vars):
            self.assertEquals(self.obj_info.variables.get(column),
                              variable)

        self.assertEquals(len(self.obj_info.primary_vars),
                          len(self.cls_info.primary_key))

    def test_checkpoint(self):
        self.obj.prop1 = 10
        self.obj_info.checkpoint()
        self.assertEquals(self.obj.prop1, 10)
        self.assertEquals(self.variable1.has_changed(), False)
        self.obj.prop1 = 20
        self.assertEquals(self.obj.prop1, 20)
        self.assertEquals(self.variable1.has_changed(), True)
        self.obj_info.checkpoint()
        self.assertEquals(self.obj.prop1, 20)
        self.assertEquals(self.variable1.has_changed(), False)
        self.obj.prop1 = 20
        self.assertEquals(self.obj.prop1, 20)
        self.assertEquals(self.variable1.has_changed(), False)

    def test_add_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.prop2 = 10
        self.obj.prop1 = 20

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

        del changes1[:]
        del changes2[:]

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

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