variables.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/>.
#
from datetime import datetime, date, time, timedelta
from decimal import Decimal
import cPickle as pickle

from storm.exceptions import NoneError
from storm.variables import *
from storm.event import EventSystem
from storm.expr import Column, SQLToken
from storm.tz import tzutc, tzoffset
from storm import Undef

from tests.helper import TestHelper


class Marker(object):
    pass

marker = Marker()


class CustomVariable(Variable):

    def __init__(self, *args, **kwargs):
        self.gets = []
        self.sets = []
        Variable.__init__(self, *args, **kwargs)

    def _parse_get(self, variable, to_db):
        self.gets.append((variable, to_db))
        return "g", variable

    def _parse_set(self, variable, from_db):
        self.sets.append((variable, from_db))
        return "s", variable


class VariableTest(TestHelper):

    def test_constructor_value(self):
        variable = CustomVariable(marker)
        self.assertEquals(variable.sets, [(marker, False)])

    def test_constructor_value_from_db(self):
        variable = CustomVariable(marker, from_db=True)
        self.assertEquals(variable.sets, [(marker, True)])

    def test_constructor_value_factory(self):
        variable = CustomVariable(value_factory=lambda:marker)
        self.assertEquals(variable.sets, [(marker, False)])

    def test_constructor_value_factory_from_db(self):
        variable = CustomVariable(value_factory=lambda:marker, from_db=True)
        self.assertEquals(variable.sets, [(marker, True)])

    def test_constructor_column(self):
        variable = CustomVariable(column=marker)
        self.assertEquals(variable.column, marker)

    def test_constructor_event(self):
        variable = CustomVariable(event=marker)
        self.assertEquals(variable.event, marker)

    def test_get_default(self):
        variable = CustomVariable()
        self.assertEquals(variable.get(default=marker), marker)

    def test_set(self):
        variable = CustomVariable()
        variable.set(marker)
        self.assertEquals(variable.sets, [(marker, False)])
        variable.set(marker, from_db=True)
        self.assertEquals(variable.sets, [(marker, False), (marker, True)])

    def test_get(self):
        variable = CustomVariable()
        variable.set(marker)
        self.assertEquals(variable.get(), ("g", ("s", marker)))
        self.assertEquals(variable.gets, [(("s", marker), False)])

        variable = CustomVariable()
        variable.set(marker)
        self.assertEquals(variable.get(to_db=True), ("g", ("s", marker)))
        self.assertEquals(variable.gets, [(("s", marker), True)])

    def test_eq(self):
        self.assertEquals(CustomVariable(marker), CustomVariable(marker))
        self.assertNotEquals(CustomVariable(marker), CustomVariable(object()))

    def test_is_defined(self):
        variable = CustomVariable()
        self.assertFalse(variable.is_defined())
        variable.set(marker)
        self.assertTrue(variable.is_defined())

    def test_set_get_none(self):
        variable = CustomVariable()
        variable.set(None)
        self.assertEquals(variable.get(marker), None)
        self.assertEquals(variable.sets, [])
        self.assertEquals(variable.gets, [])

    def test_set_none_with_allow_none(self):
        variable = CustomVariable(allow_none=False)
        self.assertRaises(NoneError, variable.set, None)

    def test_set_none_with_allow_none_and_column(self):
        column = Column("column_name")
        variable = CustomVariable(allow_none=False, column=column)
        try:
            variable.set(None)
        except NoneError, e:
            pass
        self.assertTrue("column_name" in str(e))

    def test_set_none_with_allow_none_and_column_with_table(self):
        column = Column("column_name", SQLToken("table_name"))
        variable = CustomVariable(allow_none=False, column=column)
        try:
            variable.set(None)
        except NoneError, e:
            pass
        self.assertTrue("table_name.column_name" in str(e))

    def test_event_changed(self):
        event = EventSystem(marker)

        changed_values = []
        def changed(owner, variable, old_value, new_value, fromdb):
            changed_values.append((owner, variable,
                                   old_value, new_value, fromdb))
        
        event.hook("changed", changed)

        variable = CustomVariable(event=event)
        variable.set("value1")
        variable.set("value2")
        variable.set("value3", from_db=True)
        variable.set(None, from_db=True)
        variable.set("value4")
        variable.delete()
        variable.delete()

        self.assertEquals(changed_values[0],
          (marker, variable, Undef, "value1", False))
        self.assertEquals(changed_values[1],
          (marker, variable, ("g", ("s", "value1")), "value2", False))
        self.assertEquals(changed_values[2],
          (marker, variable, ("g", ("s", "value2")), ("g", ("s", "value3")),
           True))
        self.assertEquals(changed_values[3],
          (marker, variable, ("g", ("s", "value3")), None, True))
        self.assertEquals(changed_values[4],
          (marker, variable, None, "value4", False))
        self.assertEquals(changed_values[5],
          (marker, variable, ("g", ("s", "value4")), Undef, False))
        self.assertEquals(len(changed_values), 6)

    def test_get_state(self):
        variable = CustomVariable(marker)
        self.assertEquals(variable.get_state(), (Undef, ("s", marker)))

    def test_set_state(self):
        lazy_value = object()
        variable = CustomVariable()
        variable.set_state((lazy_value, marker))
        self.assertEquals(variable.get(), ("g", marker))
        self.assertEquals(variable.get_lazy(), lazy_value)

    def test_checkpoint_and_has_changed(self):
        variable = CustomVariable()
        self.assertTrue(variable.has_changed())
        variable.set(marker)
        self.assertTrue(variable.has_changed())
        variable.checkpoint()
        self.assertFalse(variable.has_changed())
        variable.set(marker)
        self.assertFalse(variable.has_changed())
        variable.set((marker, marker))
        self.assertTrue(variable.has_changed())
        variable.checkpoint()
        self.assertFalse(variable.has_changed())
        variable.set((marker, marker))
        self.assertFalse(variable.has_changed())
        variable.set(marker)
        self.assertTrue(variable.has_changed())
        variable.set((marker, marker))
        self.assertFalse(variable.has_changed())

    def test_copy(self):
        variable = CustomVariable()
        variable.set(marker)
        variable_copy = variable.copy()
        self.assertTrue(variable is not variable_copy)
        self.assertTrue(variable == variable_copy)

    def test_hash(self):
        # They must hash the same to be used as cache keys.
        obj1 = CustomVariable(marker)
        obj2 = CustomVariable(marker)
        self.assertEquals(hash(obj1), hash(obj2))

    def test_lazy_value_setting(self):
        variable = CustomVariable()
        variable.set(LazyValue())
        self.assertEquals(variable.sets, [])
        self.assertTrue(variable.has_changed())

    def test_lazy_value_getting(self):
        variable = CustomVariable()
        variable.set(LazyValue())
        self.assertEquals(variable.get(marker), marker)
        variable.set(1)
        variable.set(LazyValue())
        self.assertEquals(variable.get(marker), marker)
        self.assertFalse(variable.is_defined())

    def test_lazy_value_resolving(self):
        event = EventSystem(marker)

        resolve_values = []
        def resolve(owner, variable, value):
            resolve_values.append((owner, variable, value))



        lazy_value = LazyValue()
        variable = CustomVariable(lazy_value, event=event)

        event.hook("resolve-lazy-value", resolve)

        variable.get()

        self.assertEquals(resolve_values,
                          [(marker, variable, lazy_value)])

    def test_lazy_value_changed_event(self):
        event = EventSystem(marker)

        changed_values = []
        def changed(owner, variable, old_value, new_value, fromdb):
            changed_values.append((owner, variable,
                                   old_value, new_value, fromdb))
        
        event.hook("changed", changed)

        variable = CustomVariable(event=event)

        lazy_value = LazyValue()

        variable.set(lazy_value)

        self.assertEquals(changed_values,
                          [(marker, variable, Undef, lazy_value, False)])

    def test_lazy_value_setting_on_resolving(self):
        event = EventSystem(marker)

        def resolve(owner, variable, value):
            variable.set(marker)

        event.hook("resolve-lazy-value", resolve)

        lazy_value = LazyValue()
        variable = CustomVariable(lazy_value, event=event)

        self.assertEquals(variable.get(), ("g", ("s", marker)))

    def test_lazy_value_reset_after_changed(self):
        event = EventSystem(marker)

        resolve_called = []
        def resolve(owner, variable, value):
            resolve_called.append(True)

        event.hook("resolve-lazy-value", resolve)

        variable = CustomVariable(event=event)

        variable.set(LazyValue())
        variable.set(1)
        self.assertEquals(variable.get(), ("g", ("s", 1)))
        self.assertEquals(resolve_called, [])

    def test_get_lazy_value(self):
        lazy_value = LazyValue()
        variable = CustomVariable()
        self.assertEquals(variable.get_lazy(), None)
        self.assertEquals(variable.get_lazy(marker), marker)
        variable.set(lazy_value)
        self.assertEquals(variable.get_lazy(marker), lazy_value)


class BoolVariableTest(TestHelper):

    def test_set_get(self):
        variable = BoolVariable()
        variable.set(1)
        self.assertTrue(variable.get() is True)
        variable.set(0)
        self.assertTrue(variable.get() is False)
        variable.set(1.1)
        self.assertTrue(variable.get() is True)
        variable.set(0.0)
        self.assertTrue(variable.get() is False)
        variable.set(Decimal(1))
        self.assertTrue(variable.get() is True)
        variable.set(Decimal(0))
        self.assertTrue(variable.get() is False)
        self.assertRaises(TypeError, variable.set, "string")


class IntVariableTest(TestHelper):

    def test_set_get(self):
        variable = IntVariable()
        variable.set(1)
        self.assertEquals(variable.get(), 1)
        variable.set(1.1)
        self.assertEquals(variable.get(), 1)
        variable.set(Decimal(2))
        self.assertEquals(variable.get(), 2)
        self.assertRaises(TypeError, variable.set, "1")


class FloatVariableTest(TestHelper):

    def test_set_get(self):
        variable = FloatVariable()
        variable.set(1.1)
        self.assertEquals(variable.get(), 1.1)
        variable.set(1)
        self.assertEquals(variable.get(), 1)
        self.assertEquals(type(variable.get()), float)
        variable.set(Decimal("1.1"))
        self.assertEquals(variable.get(), 1.1)
        self.assertRaises(TypeError, variable.set, "1")


class CharsVariableTest(TestHelper):

    def test_set_get(self):
        variable = CharsVariable()
        variable.set("str")
        self.assertEquals(variable.get(), "str")
        variable.set(buffer("buffer"))
        self.assertEquals(variable.get(), "buffer")
        self.assertRaises(TypeError, variable.set, u"unicode")


class UnicodeVariableTest(TestHelper):

    def test_set_get(self):
        variable = UnicodeVariable()
        variable.set(u"unicode")
        self.assertEquals(variable.get(), u"unicode")
        self.assertRaises(TypeError, variable.set, "str")


class DateTimeVariableTest(TestHelper):

    def test_get_set(self):
        epoch = datetime.utcfromtimestamp(0)
        variable = DateTimeVariable()
        variable.set(0)