~free.ekanayaka/storm/any-expr

#

# Copyright (c) 2006, 2007 Canonical

#

# Written by Gustavo Niemeyer <gustavo@niemeyer.net>

#

# 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

import gc

import weakref

try:

    import uuid

except ImportError:

    uuid = None

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_set_leak(self):

        """When a variable is checkpointed, the value must not leak."""

        variable = Variable()

        m = Marker()

        m_ref = weakref.ref(m)

        variable.set(m)

        variable.checkpoint()

        variable.set(LazyValue())

        del m

        gc.collect()

        self.assertIdentical(m_ref(), None)

    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_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_set_with_validator(self):

        args = []

        def validator(obj, attr, value):

            args.append((obj, attr, value))

            return value

        variable = CustomVariable(validator=validator)

        variable.set(3)

        self.assertEquals(args, [(None, None, 3)])

    def test_set_with_validator_and_validator_arguments(self):

        args = []

        def validator(obj, attr, value):

            args.append((obj, attr, value))

            return value

        variable = CustomVariable(validator=validator,

                                  validator_object_factory=lambda: 1,

                                  validator_attribute=2)

        variable.set(3)

        self.assertEquals(args, [(1, 2, 3)])

    def test_set_with_validator_raising_error(self):

        args = []

        def validator(obj, attr, value):

            args.append((obj, attr, value))

            raise ZeroDivisionError()

        variable = CustomVariable(validator=validator)

        self.assertRaises(ZeroDivisionError, variable.set, marker)

        self.assertEquals(args, [(None, None, marker)])

        self.assertEquals(variable.get(), None)

    def test_set_with_validator_changing_value(self):

        args = []

        def validator(obj, attr, value):

            args.append((obj, attr, value))

            return 42

        variable = CustomVariable(validator=validator)

        variable.set(marker)

        self.assertEquals(args, [(None, None, marker)])

        self.assertEquals(variable.get(), ('g', ('s', 42)))

    def test_set_from_db_wont_call_validator(self):

        args = []

        def validator(obj, attr, value):

            args.append((obj, attr, value))

            return 42

        variable = CustomVariable(validator=validator)

        variable.set(marker, from_db=True)

        self.assertEquals(args, [])

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

    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()

        variable_copy.gets = []

        self.assertTrue(variable is not variable_copy)

        self.assertVariablesEqual([variable], [variable_copy])

    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 DecimalVariableTest(TestHelper):

    def test_set_get(self):

        variable = DecimalVariable()

        variable.set(Decimal("1.1"))

        self.assertEquals(variable.get(), Decimal("1.1"))

        variable.set(1)

        self.assertEquals(variable.get(), 1)

        self.assertEquals(type(variable.get()), Decimal)

        variable.set(Decimal("1.1"))

        self.assertEquals(variable.get(), Decimal("1.1"))

        self.assertRaises(TypeError, variable.set, "1")

        self.assertRaises(TypeError, variable.set, 1.1)

    def test_get_set_from_database(self):

        """Strings used to/from the database."""

        variable = DecimalVariable()

        variable.set("1.1", from_db=True)

        self.assertEquals(variable.get(), Decimal("1.1"))

        self.assertEquals(variable.get(to_db=True), "1.1")

class RawStrVariableTest(TestHelper):

    def test_set_get(self):

        variable = RawStrVariable()

        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)

        self.assertEquals(variable.get(), epoch)

        variable.set(0.0)

        self.assertEquals(variable.get(), epoch)

        variable.set(0L)

        self.assertEquals(variable.get(), epoch)

        variable.set(epoch)

        self.assertEquals(variable.get(), epoch)

        self.assertRaises(TypeError, variable.set, marker)

    def test_get_set_from_database(self):

        datetime_str = "1977-05-04 12:34:56.78"

        datetime_uni = unicode(datetime_str)

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56, 780000)

        variable = DateTimeVariable()

        variable.set(datetime_str, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        variable.set(datetime_uni, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        variable.set(datetime_obj, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        datetime_str = "1977-05-04 12:34:56"

        datetime_uni = unicode(datetime_str)

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56)

        variable.set(datetime_str, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        variable.set(datetime_uni, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        variable.set(datetime_obj, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        self.assertRaises(TypeError, variable.set, 0, from_db=True)

        self.assertRaises(TypeError, variable.set, marker, from_db=True)

        self.assertRaises(ValueError, variable.set, "foobar", from_db=True)

        self.assertRaises(ValueError, variable.set, "foo bar", from_db=True)

    def test_get_set_with_tzinfo(self):

        datetime_str = "1977-05-04 12:34:56.78"

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56, 780000, tzinfo=tzutc())

        variable = DateTimeVariable(tzinfo=tzutc())

        # Naive timezone, from_db=True.

        variable.set(datetime_str, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        variable.set(datetime_obj, from_db=True)

        self.assertEquals(variable.get(), datetime_obj)

        # Naive timezone, from_db=False (doesn't work).

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56, 780000)

        self.assertRaises(ValueError, variable.set, datetime_obj)

        # Different timezone, from_db=False.

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56, 780000,

                                tzinfo=tzoffset("1h", 3600))

        variable.set(datetime_obj, from_db=False)

        converted_obj = variable.get()

        self.assertEquals(converted_obj, datetime_obj)

        self.assertEquals(type(converted_obj.tzinfo), tzutc)

        # Different timezone, from_db=True.

        datetime_obj = datetime(1977, 5, 4, 12, 34, 56, 780000,

                                tzinfo=tzoffset("1h", 3600))

        variable.set(datetime_obj, from_db=True)

        converted_obj = variable.get()

        self.assertEquals(converted_obj, datetime_obj)

        self.assertEquals(type(converted_obj.tzinfo), tzutc)

class DateVariableTest(TestHelper):

    def test_get_set(self):

        epoch = datetime.utcfromtimestamp(0)

        epoch_date = epoch.date()

        variable = DateVariable()

        variable.set(epoch)

        self.assertEquals(variable.get(), epoch_date)

        variable.set(epoch_date)

        self.assertEquals(variable.get(), epoch_date)

        self.assertRaises(TypeError, variable.set, marker)

    def test_get_set_from_database(self):

        date_str = "1977-05-04"

        date_uni = unicode(date_str)

        date_obj = date(1977, 5, 4)

        variable = DateVariable()

        variable.set(date_str, from_db=True)

        self.assertEquals(variable.get(), date_obj)

        variable.set(date_uni, from_db=True)

        self.assertEquals(variable.get(), date_obj)

        variable.set(date_obj, from_db=True)

        self.assertEquals(variable.get(), date_obj)

        self.assertRaises(TypeError, variable.set, 0, from_db=True)

        self.assertRaises(TypeError, variable.set, marker, from_db=True)

        self.assertRaises(ValueError, variable.set, "foobar", from_db=True)

    def test_set_with_datetime(self):

        datetime_str = "1977-05-04 12:34:56.78"

        date_obj = date(1977, 5, 4)

        variable = DateVariable()

        variable.set(datetime_str, from_db=True)

        self.assertEquals(variable.get(), date_obj)

class TimeVariableTest(TestHelper):

    def test_get_set(self):

        epoch = datetime.utcfromtimestamp(0)

        epoch_time = epoch.time()

        variable = TimeVariable()

        variable.set(epoch)

        self.assertEquals(variable.get(), epoch_time)

        variable.set(epoch_time)

        self.assertEquals(variable.get(), epoch_time)

        self.assertRaises(TypeError, variable.set, marker)

    def test_get_set_from_database(self):

        time_str = "12:34:56.78"

        time_uni = unicode(time_str)

        time_obj = time(12, 34, 56, 780000)

        variable = TimeVariable()

        variable.set(time_str, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        variable.set(time_uni, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        variable.set(time_obj, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        time_str = "12:34:56"

        time_uni = unicode(time_str)

        time_obj = time(12, 34, 56)

        variable.set(time_str, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        variable.set(time_uni, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        variable.set(time_obj, from_db=True)

        self.assertEquals(variable.get(), time_obj)

        self.assertRaises(TypeError, variable.set, 0, from_db=True)

        self.assertRaises(TypeError, variable.set, marker, from_db=True)