variables.py

Python的一个ORM,现在很火

        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)
        self.assertRaises(ValueError, variable.set, "foobar", from_db=True)

    def test_set_with_datetime(self):
        datetime_str = "1977-05-04 12:34:56.78"
        time_obj = time(12, 34, 56, 780000)
        variable = TimeVariable()
        variable.set(datetime_str, from_db=True)
        self.assertEquals(variable.get(), time_obj)

    def test_microsecond_error(self):
        time_str = "15:14:18.598678"
        time_obj = time(15, 14, 18, 598678)
        variable = TimeVariable()
        variable.set(time_str, from_db=True)
        self.assertEquals(variable.get(), time_obj)

    def test_microsecond_error_less_digits(self):
        time_str = "15:14:18.5986"
        time_obj = time(15, 14, 18, 598600)
        variable = TimeVariable()
        variable.set(time_str, from_db=True)
        self.assertEquals(variable.get(), time_obj)

    def test_microsecond_error_more_digits(self):
        time_str = "15:14:18.5986789"
        time_obj = time(15, 14, 18, 598678)
        variable = TimeVariable()
        variable.set(time_str, from_db=True)
        self.assertEquals(variable.get(), time_obj)


class TimeDeltaVariableTest(TestHelper):

    def test_get_set(self):
        delta = timedelta(days=42)

        variable = TimeDeltaVariable()

        variable.set(delta)
        self.assertEquals(variable.get(), delta)

        self.assertRaises(TypeError, variable.set, marker)
    
    def test_get_set_from_database(self):
        delta_str = "42 days 12:34:56.78"
        delta_uni = unicode(delta_str)
        delta_obj = timedelta(days=42, hours=12, minutes=34,
                              seconds=56, microseconds=780000)

        variable = TimeDeltaVariable()

        variable.set(delta_str, from_db=True)
        self.assertEquals(variable.get(), delta_obj)
        variable.set(delta_uni, from_db=True)
        self.assertEquals(variable.get(), delta_obj)
        variable.set(delta_obj, from_db=True)
        self.assertEquals(variable.get(), delta_obj)

        delta_str = "1 day, 12:34:56"
        delta_uni = unicode(delta_str)
        delta_obj = timedelta(days=1, hours=12, minutes=34, seconds=56)

        variable.set(delta_str, from_db=True)
        self.assertEquals(variable.get(), delta_obj)
        variable.set(delta_uni, from_db=True)
        self.assertEquals(variable.get(), delta_obj)
        variable.set(delta_obj, from_db=True)
        self.assertEquals(variable.get(), delta_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)

        # Intervals of months or years can not be converted to a
        # Python timedelta, so a ValueError exception is raised:
        self.assertRaises(ValueError, variable.set, "42 months", from_db=True)
        self.assertRaises(ValueError, variable.set, "42 years", from_db=True)


class PickleVariableTest(TestHelper):

    def test_get_set(self):
        d = {"a": 1}
        d_dump = pickle.dumps(d, -1)

        variable = PickleVariable()

        variable.set(d)
        self.assertEquals(variable.get(), d)
        self.assertEquals(variable.get(to_db=True), d_dump)

        variable.set(d_dump, from_db=True)
        self.assertEquals(variable.get(), d)
        self.assertEquals(variable.get(to_db=True), d_dump)

        self.assertEquals(variable.get_state(), (Undef, d_dump))
        
        variable.set(marker)
        variable.set_state((Undef, d_dump))
        self.assertEquals(variable.get(), d)

        variable.get()["b"] = 2
        self.assertEquals(variable.get(), {"a": 1, "b": 2})

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

        variable = PickleVariable(event=event, value_factory=list)

        changes = []
        def changed(owner, variable, old_value, new_value, fromdb):
            changes.append((variable, old_value, new_value, fromdb))

        event.hook("changed", changed)

        variable.checkpoint()

        event.emit("flush")

        self.assertEquals(changes, [])

        lst = variable.get()

        self.assertEquals(lst, [])
        self.assertEquals(changes, [])

        lst.append("a")

        self.assertEquals(changes, [])

        event.emit("flush")

        self.assertEquals(changes, [(variable, None, ["a"], False)])

        del changes[:]

        event.emit("object-deleted")
        self.assertEquals(changes, [(variable, None, ["a"], False)])


class ListVariableTest(TestHelper):

    def test_get_set(self):
        l = [1, 2]
        l_dump = pickle.dumps(l, -1)

        variable = ListVariable(IntVariable)

        variable.set(l)
        self.assertEquals(variable.get(), l)
        self.assertEquals(variable.get(to_db=True),
                          [IntVariable(1), IntVariable(2)])

        variable.set([1.1, 2.2], from_db=True)
        self.assertEquals(variable.get(), l)
        self.assertEquals(variable.get(to_db=True),
                          [IntVariable(1), IntVariable(2)])

        self.assertEquals(variable.get_state(), (Undef, l_dump))

        variable.set([])
        variable.set_state((Undef, l_dump))
        self.assertEquals(variable.get(), l)

        variable.get().append(3)
        self.assertEquals(variable.get(), [1, 2, 3])

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

        variable = ListVariable(CharsVariable, event=event,
                                value_factory=list)

        changes = []
        def changed(owner, variable, old_value, new_value, fromdb):
            changes.append((variable, old_value, new_value, fromdb))

        event.hook("changed", changed)

        variable.checkpoint()

        event.emit("flush")

        self.assertEquals(changes, [])

        lst = variable.get()

        self.assertEquals(lst, [])
        self.assertEquals(changes, [])

        lst.append("a")

        self.assertEquals(changes, [])

        event.emit("flush")

        self.assertEquals(changes, [(variable, None, ["a"], False)])

        del changes[:]

        event.emit("object-deleted")
        self.assertEquals(changes, [(variable, None, ["a"], False)])


class EnumVariableTest(TestHelper):

    def test_set_get(self):
        variable = EnumVariable({1: "foo", 2: "bar"}, {"foo": 1, "bar": 2})
        variable.set("foo")
        self.assertEquals(variable.get(), "foo")
        self.assertEquals(variable.get(to_db=True), 1)
        variable.set(2, from_db=True)
        self.assertEquals(variable.get(), "bar")
        self.assertEquals(variable.get(to_db=True), 2)
        self.assertRaises(ValueError, variable.set, "foobar")
        self.assertRaises(ValueError, variable.set, 2)

    def test_in_map(self):
        variable = EnumVariable({1: "foo", 2: "bar"}, {"one": 1, "two": 2})
        variable.set("one")
        self.assertEquals(variable.get(), "foo")
        self.assertEquals(variable.get(to_db=True), 1)
        variable.set(2, from_db=True)
        self.assertEquals(variable.get(), "bar")
        self.assertEquals(variable.get(to_db=True), 2)
        self.assertRaises(ValueError, variable.set, "foo")
        self.assertRaises(ValueError, variable.set, 2)