sqlobject.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 datetime

from storm.database import create_database
from storm.exceptions import NoneError
from storm.sqlobject import *
from storm.store import Store
from storm.expr import Asc, Like
from storm.tz import tzutc

from tests.helper import TestHelper


class SQLObjectTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)

        # Allow classes with the same name in different tests to resolve
        # property path strings properly.
        SQLObjectBase._storm_property_registry.clear()

        self.store = Store(create_database("sqlite:"))
        class SQLObject(SQLObjectBase):
            @staticmethod
            def _get_store():
                return self.store

        self.SQLObject = SQLObject

        self.store.execute("CREATE TABLE person "
                           "(id INTEGER PRIMARY KEY, name TEXT, age INTEGER,"
                           " ts TIMESTAMP, delta INTERVAL,"
                           " address_id INTEGER)")
        self.store.execute("INSERT INTO person VALUES "
                           "(1, 'John Joe', 20, '2007-02-05 19:53:15',"
                           " '1 day, 12:34:56', 1)")
        self.store.execute("INSERT INTO person VALUES "
                           "(2, 'John Doe', 20, '2007-02-05 20:53:15',"
                           " '42 days 12:34:56.78', 2)")

        self.store.execute("CREATE TABLE address "
                           "(id INTEGER PRIMARY KEY, city TEXT)")
        self.store.execute("INSERT INTO address VALUES (1, 'Curitiba')")
        self.store.execute("INSERT INTO address VALUES (2, 'Sao Carlos')")

        self.store.execute("CREATE TABLE phone "
                           "(id INTEGER PRIMARY KEY, person_id INTEGER,"
                           "number TEXT)")
        self.store.execute("INSERT INTO phone VALUES (1, 2, '1234-5678')")
        self.store.execute("INSERT INTO phone VALUES (2, 1, '8765-4321')")
        self.store.execute("INSERT INTO phone VALUES (3, 2, '8765-5678')")

        self.store.execute("CREATE TABLE person_phone "
                           "(person_id INTEGER , phone_id INTEGER)")
        self.store.execute("INSERT INTO person_phone VALUES (2, 1)")
        self.store.execute("INSERT INTO person_phone VALUES (2, 2)")
        self.store.execute("INSERT INTO person_phone VALUES (1, 1)")

        class Person(self.SQLObject):
            _defaultOrder = "-Person.name"
            name = StringCol()
            age = IntCol()
            ts = UtcDateTimeCol()

        self.Person = Person

    def test_get(self):
        person = self.Person.get(2)
        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_get_not_found(self):
        self.assertRaises(SQLObjectNotFound, self.Person.get, 1000)

    def test_destroySelf(self):
        person = self.Person.get(2)
        person.destroySelf()
        self.assertRaises(SQLObjectNotFound, self.Person.get, 2)

    def test_delete(self):
        self.Person.delete(2)
        self.assertRaises(SQLObjectNotFound, self.Person.get, 2)

    def test_custom_table_name(self):
        class MyPerson(self.Person):
            _table = "person"

        person = MyPerson.get(2)

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_custom_id_name(self):
        class MyPerson(self.SQLObject):
            _defaultOrder = "-Person.name"
            _table = "person"
            _idName = "name"
            _idType = unicode
            age = IntCol()
            ts = UtcDateTimeCol()

        person = MyPerson.get("John Doe")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_create(self):
        person = self.Person(name="John Joe")

        self.store.flush()

        self.assertTrue(Store.of(person) is self.store)
        self.assertEquals(type(person.id), int)
        self.assertEquals(person.name, "John Joe")

    def test_init_hook(self):
        called = []
        class Person(self.Person):
            def _init(self, *args, **kwargs):
                called.append(True)

        person = Person(name="John Joe")
        self.assertEquals(called, [True])
        Person.get(2)
        self.assertEquals(called, [True, True])

    def test_alternateID(self):
        class Person(self.SQLObject):
            name = StringCol(alternateID=True)
        person = Person.byName("John Doe")
        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_alternateMethodName(self):
        class Person(self.SQLObject):
            name = StringCol(alternateMethodName="byFoo")

        person = Person.byFoo("John Doe")
        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

        self.assertRaises(SQLObjectNotFound, Person.byFoo, "John None")

    def test_select(self):
        result = self.Person.select("name = 'John Joe'")
        self.assertEquals(result[0].name, "John Joe")

    def test_select_sqlbuilder(self):
        result = self.Person.select(self.Person.q.name == 'John Joe')
        self.assertEqual(result[0].name, "John Joe")

    def test_select_orderBy(self):
        result = self.Person.select("name LIKE 'John%'", orderBy=("name","id"))
        self.assertEquals(result[0].name, "John Doe")

    def test_select_orderBy_expr(self):
        result = self.Person.select("name LIKE 'John%'",
                                    orderBy=self.Person.name)
        self.assertEquals(result[0].name, "John Doe")

    def test_select_all(self):
        result = self.Person.select()
        self.assertEquals(result[0].name, "John Joe")

    def test_select_limit(self):
        result = self.Person.select(limit=1)
        self.assertEquals(len(list(result)), 1)

    def test_select_distinct(self):
        result = self.Person.select("person.name = 'John Joe'",
                                    clauseTables=["phone"], distinct=True)
        self.assertEquals(len(list(result)), 1)

    def test_select_clauseTables_simple(self):
        result = self.Person.select("name = 'John Joe'", ["person"])
        self.assertEquals(result[0].name, "John Joe")

    def test_select_clauseTables_implicit_join(self):
        result = self.Person.select("person.name = 'John Joe' and "
                                    "phone.person_id = person.id",
                                    ["Person", "phone"])
        self.assertEquals(result[0].name, "John Joe")

    def test_select_clauseTables_no_cls_table(self):
        result = self.Person.select("person.name = 'John Joe' and "
                                    "phone.person_id = person.id",
                                    ["phone"])
        self.assertEquals(result[0].name, "John Joe")

    def test_selectBy(self):
        result = self.Person.selectBy(name="John Joe")
        self.assertEquals(result[0].name, "John Joe")

    def test_selectBy_orderBy(self):
        result = self.Person.selectBy(age=20, orderBy="name")
        self.assertEquals(result[0].name, "John Doe")

        result = self.Person.selectBy(age=20, orderBy="-name")
        self.assertEquals(result[0].name, "John Joe")

    def test_selectOne(self):
        person = self.Person.selectOne("name = 'John Joe'")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Joe")

        nobody = self.Person.selectOne("name = 'John None'")

        self.assertEquals(nobody, None)

        # SQLBuilder style expression:
        person = self.Person.selectOne(self.Person.q.name == 'John Joe')

        self.assertNotEqual(person, None)
        self.assertEqual(person.name, 'John Joe')

    def test_selectOne_clauseTables(self):
        person = self.Person.selectOne("person.name = 'John Joe' and "
                                       "phone.person_id = person.id",
                                       ["phone"])
        self.assertEquals(person.name, "John Joe")

    def test_selectOneBy(self):
        person = self.Person.selectOneBy(name="John Joe")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Joe")

        nobody = self.Person.selectOneBy(name="John None")

        self.assertEquals(nobody, None)

    def test_selectFirst(self):
        person = self.Person.selectFirst("name LIKE 'John%'", orderBy="name")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

        person = self.Person.selectFirst("name LIKE 'John%'", orderBy="-name")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Joe")

        nobody = self.Person.selectFirst("name = 'John None'", orderBy="name")

        self.assertEquals(nobody, None)

        # SQLBuilder style expression:
        person = self.Person.selectFirst(LIKE(self.Person.q.name, 'John%'),
                                         orderBy="name")
        self.assertNotEqual(person, None)
        self.assertEqual(person.name, 'John Doe')

    def test_selectFirst_default_order(self):
        person = self.Person.selectFirst("name LIKE 'John%'")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Joe")

    def test_selectFirst_default_order_list(self):
        class Person(self.Person):
            _defaultOrder = ["name"]

        person = Person.selectFirst("name LIKE 'John%'")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_selectFirst_default_order_expr(self):
        class Person(self.Person):
            _defaultOrder = [SQLConstant("name")]

        person = Person.selectFirst("name LIKE 'John%'")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_selectFirst_default_order_fully_qualified(self):
        class Person(self.Person):
            _defaultOrder = ["person.name"]

        person = Person.selectFirst("name LIKE 'John%'")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")

    def test_selectFirstBy(self):
        person = self.Person.selectFirstBy(age=20, orderBy="name")

        self.assertTrue(person)
        self.assertEquals(person.name, "John Doe")