expr.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 tests.helper import TestHelper

from storm.variables import *
from storm.expr import *


class Func1(NamedFunc):
    name = "func1"

class Func2(NamedFunc):
    name = "func2"

# Create columnN, tableN, and elemN variables.
for i in range(10):
    for name in ["column", "elem"]:
        exec "%s%d = SQLRaw('%s%d')" % (name, i, name, i)
    for name in ["table"]:
        exec "%s%d = '%s%d'" % (name, i, name, i)


class TrackContext(FromExpr):
    context = None

@compile.when(TrackContext)
def compile_track_context(compile, state, expr):
    expr.context = state.context
    return ""

def track_contexts(n):
    return [TrackContext() for i in range(n)]


class ExprTest(TestHelper):

    def test_select_default(self):
        expr = Select(())
        self.assertEquals(expr.columns, ())
        self.assertEquals(expr.where, Undef)
        self.assertEquals(expr.tables, Undef)
        self.assertEquals(expr.default_tables, Undef)
        self.assertEquals(expr.order_by, Undef)
        self.assertEquals(expr.group_by, Undef)
        self.assertEquals(expr.limit, Undef)
        self.assertEquals(expr.offset, Undef)
        self.assertEquals(expr.distinct, False)

    def test_select_constructor(self):
        objects = [object() for i in range(9)]
        expr = Select(*objects)
        self.assertEquals(expr.columns, objects[0])
        self.assertEquals(expr.where, objects[1])
        self.assertEquals(expr.tables, objects[2])
        self.assertEquals(expr.default_tables, objects[3])
        self.assertEquals(expr.order_by, objects[4])
        self.assertEquals(expr.group_by, objects[5])
        self.assertEquals(expr.limit, objects[6])
        self.assertEquals(expr.offset, objects[7])
        self.assertEquals(expr.distinct, objects[8])

    def test_insert_default(self):
        expr = Insert(None, None)
        self.assertEquals(expr.columns, None)
        self.assertEquals(expr.values, None)
        self.assertEquals(expr.table, Undef)
        self.assertEquals(expr.default_table, Undef)

    def test_insert_constructor(self):
        objects = [object() for i in range(4)]
        expr = Insert(*objects)
        self.assertEquals(expr.columns, objects[0])
        self.assertEquals(expr.values, objects[1])
        self.assertEquals(expr.table, objects[2])
        self.assertEquals(expr.default_table, objects[3])

    def test_update_default(self):
        expr = Update(None)
        self.assertEquals(expr.set, None)
        self.assertEquals(expr.where, Undef)
        self.assertEquals(expr.table, Undef)
        self.assertEquals(expr.default_table, Undef)

    def test_update_constructor(self):
        objects = [object() for i in range(4)]
        expr = Update(*objects)
        self.assertEquals(expr.set, objects[0])
        self.assertEquals(expr.where, objects[1])
        self.assertEquals(expr.table, objects[2])
        self.assertEquals(expr.default_table, objects[3])

    def test_delete_default(self):
        expr = Delete()
        self.assertEquals(expr.where, Undef)
        self.assertEquals(expr.table, Undef)

    def test_delete_constructor(self):
        objects = [object() for i in range(3)]
        expr = Delete(*objects)
        self.assertEquals(expr.where, objects[0])
        self.assertEquals(expr.table, objects[1])
        self.assertEquals(expr.default_table, objects[2])

    def test_and(self):
        expr = And(elem1, elem2, elem3)
        self.assertEquals(expr.exprs, (elem1, elem2, elem3))

    def test_or(self):
        expr = Or(elem1, elem2, elem3)
        self.assertEquals(expr.exprs, (elem1, elem2, elem3))

    def test_column_default(self):
        expr = Column()
        self.assertEquals(expr.name, Undef)
        self.assertEquals(expr.table, Undef)

        # Test for identity. We don't want False there.
        self.assertTrue(expr.primary is 0)

        self.assertEquals(expr.variable_factory, Variable)

    def test_column_constructor(self):
        objects = [object() for i in range(3)]
        objects.insert(2, True)
        expr = Column(*objects)
        self.assertEquals(expr.name, objects[0])
        self.assertEquals(expr.table, objects[1])

        # Test for identity. We don't want True there either.
        self.assertTrue(expr.primary is 1)

        self.assertEquals(expr.variable_factory, objects[3])

    def test_func(self):
        expr = Func("myfunc", elem1, elem2)
        self.assertEquals(expr.name, "myfunc")
        self.assertEquals(expr.args, (elem1, elem2))

    def test_named_func(self):
        class MyFunc(NamedFunc):
            name = "myfunc"
        expr = MyFunc(elem1, elem2)
        self.assertEquals(expr.name, "myfunc")
        self.assertEquals(expr.args, (elem1, elem2))

    def test_like(self):
        expr = Like(elem1, elem2)
        self.assertEquals(expr.expr1, elem1)
        self.assertEquals(expr.expr2, elem2)

    def test_like(self):
        expr = Like(elem1, elem2, elem3)
        self.assertEquals(expr.expr1, elem1)
        self.assertEquals(expr.expr2, elem2)
        self.assertEquals(expr.escape, elem3)

    def test_eq(self):
        expr = Eq(elem1, elem2)
        self.assertEquals(expr.expr1, elem1)
        self.assertEquals(expr.expr2, elem2)

    def test_sql_default(self):
        expr = SQL(None)
        self.assertEquals(expr.expr, None)
        self.assertEquals(expr.params, Undef)
        self.assertEquals(expr.tables, Undef)

    def test_sql_constructor(self):
        objects = [object() for i in range(3)]
        expr = SQL(*objects)
        self.assertEquals(expr.expr, objects[0])
        self.assertEquals(expr.params, objects[1])
        self.assertEquals(expr.tables, objects[2])

    def test_join_expr_right(self):
        expr = JoinExpr(None)
        self.assertEquals(expr.right, None)
        self.assertEquals(expr.left, Undef)
        self.assertEquals(expr.on, Undef)

    def test_join_expr_on(self):
        on = Expr()
        expr = JoinExpr(None, on)
        self.assertEquals(expr.right, None)
        self.assertEquals(expr.left, Undef)
        self.assertEquals(expr.on, on)

    def test_join_expr_on_keyword(self):
        on = Expr()
        expr = JoinExpr(None, on=on)
        self.assertEquals(expr.right, None)
        self.assertEquals(expr.left, Undef)
        self.assertEquals(expr.on, on)

    def test_join_expr_on_invalid(self):
        on = Expr()
        self.assertRaises(ExprError, JoinExpr, None, on, None)

    def test_join_expr_right_left(self):
        objects = [object() for i in range(2)]
        expr = JoinExpr(*objects)
        self.assertEquals(expr.left, objects[0])
        self.assertEquals(expr.right, objects[1])
        self.assertEquals(expr.on, Undef)

    def test_join_expr_right_left_on(self):
        objects = [object() for i in range(3)]
        expr = JoinExpr(*objects)
        self.assertEquals(expr.left, objects[0])
        self.assertEquals(expr.right, objects[1])
        self.assertEquals(expr.on, objects[2])

    def test_join_expr_right_join(self):
        join = JoinExpr(None)
        expr = JoinExpr(None, join)
        self.assertEquals(expr.right, join)
        self.assertEquals(expr.left, None)
        self.assertEquals(expr.on, Undef)

    def test_table(self):
        objects = [object() for i in range(1)]
        expr = Table(*objects)
        self.assertEquals(expr.name, objects[0])

    def test_alias_default(self):
        expr = Alias(None)
        self.assertEquals(expr.expr, None)
        self.assertTrue(isinstance(expr.name, str))

    def test_alias_constructor(self):
        objects = [object() for i in range(2)]
        expr = Alias(*objects)
        self.assertEquals(expr.expr, objects[0])
        self.assertEquals(expr.name, objects[1])

    def test_union(self):
        expr = Union(elem1, elem2, elem3)
        self.assertEquals(expr.exprs, (elem1, elem2, elem3))

    def test_union_with_kwargs(self):
        expr = Union(elem1, elem2, all=True, order_by=(), limit=1, offset=2)
        self.assertEquals(expr.exprs, (elem1, elem2))
        self.assertEquals(expr.all, True)
        self.assertEquals(expr.order_by, ())
        self.assertEquals(expr.limit, 1)
        self.assertEquals(expr.offset, 2)

    def test_except(self):
        expr = Except(elem1, elem2, elem3)
        self.assertEquals(expr.exprs, (elem1, elem2, elem3))

    def test_except_with_kwargs(self):
        expr = Except(elem1, elem2, all=True, order_by=(), limit=1, offset=2)
        self.assertEquals(expr.exprs, (elem1, elem2))
        self.assertEquals(expr.all, True)
        self.assertEquals(expr.order_by, ())
        self.assertEquals(expr.limit, 1)
        self.assertEquals(expr.offset, 2)

    def test_intersect(self):
        expr = Intersect(elem1, elem2, elem3)
        self.assertEquals(expr.exprs, (elem1, elem2, elem3))

    def test_intersect_with_kwargs(self):
        expr = Intersect(elem1, elem2, all=True, order_by=(), limit=1, offset=2)
        self.assertEquals(expr.exprs, (elem1, elem2))
        self.assertEquals(expr.all, True)
        self.assertEquals(expr.order_by, ())
        self.assertEquals(expr.limit, 1)
        self.assertEquals(expr.offset, 2)

    def test_auto_table(self):
        expr = AutoTable(elem1, elem2)
        self.assertEquals(expr.expr, elem1)
        self.assertEquals(expr.table, elem2)


class StateTest(TestHelper):

    def setUp(self):
        TestHelper.setUp(self)
        self.state = State()

    def test_attrs(self):
        self.assertEquals(self.state.parameters, [])
        self.assertEquals(self.state.auto_tables, [])
        self.assertEquals(self.state.context, None)

    def test_push_pop(self):
        self.state.parameters.extend([1, 2])
        self.state.push("parameters", [])
        self.assertEquals(self.state.parameters, [])
        self.state.pop()
        self.assertEquals(self.state.parameters, [1, 2])
        self.state.push("parameters")
        self.assertEquals(self.state.parameters, [1, 2])
        self.state.parameters.append(3)
        self.assertEquals(self.state.parameters, [1, 2, 3])
        self.state.pop()
        self.assertEquals(self.state.parameters, [1, 2])

    def test_push_pop_unexistent(self):
        self.state.push("nonexistent")
        self.assertEquals(self.state.nonexistent, None)
        self.state.nonexistent = "something"
        self.state.pop()
        self.assertEquals(self.state.nonexistent, None)


class CompileTest(TestHelper):

    def test_customize(self):
        custom_compile = compile.fork()
        @custom_compile.when(type(None))
        def compile_none(compile, state, expr):
            return "None"
        statement, parameters = custom_compile(Func1(None))
        self.assertEquals(statement, "func1(None)")

    def test_customize_inheritance(self):
        class C(object): pass
        compile_parent = Compile()
        compile_child = compile_parent.fork()