expr.py

Python的一个ORM,现在很火

        self.assertEquals(statement, "NATURAL RIGHT JOIN func1() "
                                     "ON func2() = ?")
        self.assertEquals(parameters, [Variable("value")])

    def test_union(self):
        expr = Union(Func1(), elem2, elem3)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() UNION elem2 UNION elem3")
        self.assertEquals(parameters, [])

    def test_union_all(self):
        expr = Union(Func1(), elem2, elem3, all=True)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() UNION ALL elem2 UNION ALL elem3")
        self.assertEquals(parameters, [])

    def test_union_order_by_limit_offset(self):
        expr = Union(elem1, elem2, order_by=Func1(), limit=1, offset=2)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "elem1 UNION elem2 ORDER BY func1() "
                                     "LIMIT 1 OFFSET 2")
        self.assertEquals(parameters, [])

    def test_union_select(self):
        expr = Union(Select(elem1), Select(elem2))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) UNION (SELECT elem2)")
        self.assertEquals(parameters, [])

    def test_union_select_nested(self):
        expr = Union(Select(elem1), Union(Select(elem2), Select(elem3)))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) UNION"
                                     " ((SELECT elem2) UNION (SELECT elem3))")
        self.assertEquals(parameters, [])

    def test_union_contexts(self):
        select1, select2, order_by = track_contexts(3)
        expr = Union(select1, select2, order_by=order_by)
        compile(expr)
        self.assertEquals(select1.context, SELECT)
        self.assertEquals(select2.context, SELECT)
        self.assertEquals(order_by.context, COLUMN_NAME)

    def test_except(self):
        expr = Except(Func1(), elem2, elem3)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() EXCEPT elem2 EXCEPT elem3")
        self.assertEquals(parameters, [])

    def test_except_all(self):
        expr = Except(Func1(), elem2, elem3, all=True)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() EXCEPT ALL elem2 "
                                     "EXCEPT ALL elem3")
        self.assertEquals(parameters, [])

    def test_except_order_by_limit_offset(self):
        expr = Except(elem1, elem2, order_by=Func1(), limit=1, offset=2)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "elem1 EXCEPT elem2 ORDER BY func1() "
                                     "LIMIT 1 OFFSET 2")
        self.assertEquals(parameters, [])

    def test_except_select(self):
        expr = Except(Select(elem1), Select(elem2))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) EXCEPT (SELECT elem2)")
        self.assertEquals(parameters, [])

    def test_except_select_nested(self):
        expr = Except(Select(elem1), Except(Select(elem2), Select(elem3)))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) EXCEPT"
                                     " ((SELECT elem2) EXCEPT (SELECT elem3))")
        self.assertEquals(parameters, [])

    def test_except_contexts(self):
        select1, select2, order_by = track_contexts(3)
        expr = Except(select1, select2, order_by=order_by)
        compile(expr)
        self.assertEquals(select1.context, SELECT)
        self.assertEquals(select2.context, SELECT)
        self.assertEquals(order_by.context, COLUMN_NAME)

    def test_intersect(self):
        expr = Intersect(Func1(), elem2, elem3)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() INTERSECT elem2 INTERSECT elem3")
        self.assertEquals(parameters, [])

    def test_intersect_all(self):
        expr = Intersect(Func1(), elem2, elem3, all=True)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "func1() INTERSECT ALL elem2 INTERSECT ALL elem3")
        self.assertEquals(parameters, [])

    def test_intersect_order_by_limit_offset(self):
        expr = Intersect(elem1, elem2, order_by=Func1(), limit=1, offset=2)
        statement, parameters = compile(expr)
        self.assertEquals(statement, "elem1 INTERSECT elem2 ORDER BY func1() "
                                     "LIMIT 1 OFFSET 2")
        self.assertEquals(parameters, [])

    def test_intersect_select(self):
        expr = Intersect(Select(elem1), Select(elem2))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) INTERSECT (SELECT elem2)")
        self.assertEquals(parameters, [])

    def test_intersect_select_nested(self):
        expr = Intersect(Select(elem1), Intersect(Select(elem2), Select(elem3)))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "(SELECT elem1) INTERSECT"
                                     " ((SELECT elem2) INTERSECT (SELECT elem3))")
        self.assertEquals(parameters, [])

    def test_intersect_contexts(self):
        select1, select2, order_by = track_contexts(3)
        expr = Intersect(select1, select2, order_by=order_by)
        compile(expr)
        self.assertEquals(select1.context, SELECT)
        self.assertEquals(select2.context, SELECT)
        self.assertEquals(order_by.context, COLUMN_NAME)

    def test_auto_table(self):
        expr = Select(AutoTable(1, table1))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT ? FROM table1")
        self.assertEquals(parameters, [IntVariable(1)])

    def test_auto_table_with_column(self):
        expr = Select(AutoTable(Column(elem1, table1), table2))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT table1.elem1 FROM table1, table2")
        self.assertEquals(parameters, [])

    def test_auto_table_with_column_and_replace(self):
        expr = Select(AutoTable(Column(elem1, table1), table2, replace=True))
        statement, parameters = compile(expr)
        self.assertEquals(statement, "SELECT table1.elem1 FROM table2")
        self.assertEquals(parameters, [])

    def test_auto_table_with_join(self):
        expr = Select(AutoTable(Column(elem1, table1), LeftJoin(table2)))
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT table1.elem1 FROM table1 LEFT JOIN table2")
        self.assertEquals(parameters, [])

    def test_auto_table_with_join_with_left_table(self):
        expr = Select(AutoTable(Column(elem1, table1),
                                LeftJoin(table1, table2)))
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT table1.elem1 FROM table1 LEFT JOIN table2")
        self.assertEquals(parameters, [])

    def test_auto_table_duplicated(self):
        expr = Select([AutoTable(Column(elem1, table1), Join(table2)),
                       AutoTable(Column(elem2, table2), Join(table1)),
                       AutoTable(Column(elem3, table1), Join(table1)),
                       AutoTable(Column(elem4, table3), table1),
                       AutoTable(Column(elem5, table1), Join(table4, table5))])
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT table1.elem1, table2.elem2, table1.elem3,"
                                " table3.elem4, table1.elem5 "
                          "FROM table3,"
                              " table4 JOIN table5 JOIN table1 JOIN table2")
        self.assertEquals(parameters, [])

    def test_auto_table_duplicated_nested(self):
        expr = Select(AutoTable(Column(elem1, table1), Join(table2)),
                      In(1, Select(AutoTable(Column(elem1, table1),
                                             Join(table2)))))
        statement, parameters = compile(expr)
        self.assertEquals(statement,
                          "SELECT table1.elem1 FROM table1 JOIN table2 "
                          "WHERE ? IN "
                          "(SELECT table1.elem1 FROM table1 JOIN table2)")
        self.assertEquals(parameters, [IntVariable(1)])



class CompilePythonTest(TestHelper):

    def test_precedence(self):
        for i in range(10):
            exec "e%d = SQLRaw('%d')" % (i, i)
        expr = And(e1, Or(e2, e3),
                   Add(e4, Mul(e5, Sub(e6, Div(e7, Div(e8, e9))))))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 and (2 or 3) and 4+5*(6-7/(8/9))")

    def test_get_precedence(self):
        self.assertTrue(compile_python.get_precedence(Or) <
                        compile_python.get_precedence(And))
        self.assertTrue(compile_python.get_precedence(Add) <
                        compile_python.get_precedence(Mul))
        self.assertTrue(compile_python.get_precedence(Sub) <
                        compile_python.get_precedence(Div))

    def test_compile_sequence(self):
        expr = [elem1, Variable(1), (Variable(2), None)]
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "elem1, 1, 2, None")

    def test_compile_invalid(self):
        self.assertRaises(CompileError, compile_python, object())
        self.assertRaises(CompileError, compile_python, [object()])

    def test_compile_unsupported(self):
        self.assertRaises(CompileError, compile_python, Expr())
        self.assertRaises(CompileError, compile_python, Func1())

    def test_str(self):
        py_expr = compile_python.get_expr("str")
        self.assertEquals(py_expr, "'str'")

    def test_unicode(self):
        py_expr = compile_python.get_expr(u"str")
        self.assertEquals(py_expr, "u'str'")

    def test_int(self):
        py_expr = compile_python.get_expr(1)
        self.assertEquals(py_expr, "1")

    def test_long(self):
        py_expr = compile_python.get_expr(1L)
        self.assertEquals(py_expr, "1L")

    def test_bool(self):
        py_expr = compile_python.get_expr(True)
        self.assertEquals(py_expr, "True")

    def test_float(self):
        py_expr = compile_python.get_expr(1.1)
        self.assertEquals(py_expr, repr(1.1))

    def test_datetime(self):
        dt = datetime(1977, 5, 4, 12, 34)
        py_expr = compile_python.get_expr(dt)
        self.assertEquals(py_expr, repr(dt))

    def test_date(self):
        d = date(1977, 5, 4)
        py_expr = compile_python.get_expr(d)
        self.assertEquals(py_expr, repr(d))

    def test_time(self):
        t = time(12, 34)
        py_expr = compile_python.get_expr(t)
        self.assertEquals(py_expr, repr(t))

    def test_timedelta(self):
        td = timedelta(days=1, seconds=2, microseconds=3)
        py_expr = compile_python.get_expr(td)
        self.assertEquals(py_expr, repr(td))

    def test_none(self):
        py_expr = compile_python.get_expr(None)
        self.assertEquals(py_expr, "None")

    def test_column(self):
        expr = Column(column1)
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "get_column('column1')")

    def test_column_table(self):
        expr = Column(column1, table1)
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "get_column('column1')")

    def test_variable(self):
        expr = Variable("value")
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "'value'")

    def test_eq(self):
        expr = Eq(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 == 2")

    def test_ne(self):
        expr = Ne(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 != 2")

    def test_gt(self):
        expr = Gt(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 > 2")

    def test_ge(self):
        expr = Ge(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 >= 2")

    def test_lt(self):
        expr = Lt(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 < 2")

    def test_le(self):
        expr = Le(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 <= 2")

    def test_lshift(self):
        expr = LShift(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1<<2")

    def test_rshift(self):
        expr = RShift(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1>>2")

    def test_in(self):
        expr = In(Variable(1), Variable(2))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "1 in (2,)")

    def test_and(self):
        expr = And(elem1, elem2, And(elem3, elem4))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "elem1 and elem2 and elem3 and elem4")

    def test_or(self):
        expr = Or(elem1, elem2, Or(elem3, elem4))
        py_expr = compile_python.get_expr(expr)
        self.assertEquals(py_expr, "elem1 or elem2 or elem3 or elem4")

    def test_add(self):
        expr = Add(elem1, elem2, Add(elem3, elem4))
        py_expr = compile_python.g