from sqlalchemy.testing import fixtures, eq_, is_, is_not_ from sqlalchemy import testing from sqlalchemy.testing import assert_raises_message from sqlalchemy.sql import column, desc, asc, literal, collate, null, true, false from sqlalchemy.sql.expression import BinaryExpression, \ ClauseList, Grouping, \ UnaryExpression, select, union, func, tuple_ from sqlalchemy.sql import operators, table import operator from sqlalchemy import String, Integer, LargeBinary from sqlalchemy import exc from sqlalchemy.engine import default from sqlalchemy.sql.elements import _literal_as_text from sqlalchemy.schema import Column, Table, MetaData from sqlalchemy.types import TypeEngine, TypeDecorator, UserDefinedType, \ Boolean, NullType, MatchType from sqlalchemy.dialects import mysql, firebird, postgresql, oracle, \ sqlite, mssql from sqlalchemy import util import datetime import collections from sqlalchemy import text, literal_column from sqlalchemy import and_, not_, between, or_ from sqlalchemy.sql import true, false, null class LoopOperate(operators.ColumnOperators): def operate(self, op, *other, **kwargs): return op class DefaultColumnComparatorTest(fixtures.TestBase): def _do_scalar_test(self, operator, compare_to): left = column('left') assert left.comparator.operate(operator).compare( compare_to(left) ) self._loop_test(operator) def _do_operate_test(self, operator, right=column('right')): left = column('left') assert left.comparator.operate( operator, right).compare( BinaryExpression( _literal_as_text(left), _literal_as_text(right), operator)) assert operator( left, right).compare( BinaryExpression( _literal_as_text(left), _literal_as_text(right), operator)) self._loop_test(operator, right) def _loop_test(self, operator, *arg): l = LoopOperate() is_( operator(l, *arg), operator ) def test_desc(self): self._do_scalar_test(operators.desc_op, desc) def test_asc(self): self._do_scalar_test(operators.asc_op, asc) def test_plus(self): self._do_operate_test(operators.add) def test_is_null(self): self._do_operate_test(operators.is_, None) def test_isnot_null(self): self._do_operate_test(operators.isnot, None) def test_is_null_const(self): self._do_operate_test(operators.is_, null()) def test_is_true_const(self): self._do_operate_test(operators.is_, true()) def test_is_false_const(self): self._do_operate_test(operators.is_, false()) def test_equals_true(self): self._do_operate_test(operators.eq, True) def test_notequals_true(self): self._do_operate_test(operators.ne, True) def test_is_true(self): self._do_operate_test(operators.is_, True) def test_isnot_true(self): self._do_operate_test(operators.isnot, True) def test_is_false(self): self._do_operate_test(operators.is_, False) def test_isnot_false(self): self._do_operate_test(operators.isnot, False) def test_like(self): self._do_operate_test(operators.like_op) def test_notlike(self): self._do_operate_test(operators.notlike_op) def test_ilike(self): self._do_operate_test(operators.ilike_op) def test_notilike(self): self._do_operate_test(operators.notilike_op) def test_is(self): self._do_operate_test(operators.is_) def test_isnot(self): self._do_operate_test(operators.isnot) def test_no_getitem(self): assert_raises_message( NotImplementedError, "Operator 'getitem' is not supported on this expression", self._do_operate_test, operators.getitem ) assert_raises_message( NotImplementedError, "Operator 'getitem' is not supported on this expression", lambda: column('left')[3] ) def test_in(self): left = column('left') assert left.comparator.operate(operators.in_op, [1, 2, 3]).compare( BinaryExpression( left, Grouping(ClauseList( literal(1), literal(2), literal(3) )), operators.in_op ) ) self._loop_test(operators.in_op, [1, 2, 3]) def test_notin(self): left = column('left') assert left.comparator.operate(operators.notin_op, [1, 2, 3]).compare( BinaryExpression( left, Grouping(ClauseList( literal(1), literal(2), literal(3) )), operators.notin_op ) ) self._loop_test(operators.notin_op, [1, 2, 3]) def test_in_no_accept_list_of_non_column_element(self): left = column('left') foo = ClauseList() assert_raises_message( exc.InvalidRequestError, r"in_\(\) accepts either a list of expressions or a selectable:", left.in_, [foo] ) def test_in_no_accept_non_list_non_selectable(self): left = column('left') right = column('right') assert_raises_message( exc.InvalidRequestError, r"in_\(\) accepts either a list of expressions or a selectable:", left.in_, right ) def test_in_no_accept_non_list_thing_with_getitem(self): # test [ticket:2726] class HasGetitem(String): class comparator_factory(String.Comparator): def __getitem__(self, value): return value left = column('left') right = column('right', HasGetitem) assert_raises_message( exc.InvalidRequestError, r"in_\(\) accepts either a list of expressions or a selectable:", left.in_, right ) def test_collate(self): left = column('left') right = "some collation" left.comparator.operate(operators.collate, right).compare( collate(left, right) ) def test_concat(self): self._do_operate_test(operators.concat_op) class CustomUnaryOperatorTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' def _factorial_fixture(self): class MyInteger(Integer): class comparator_factory(Integer.Comparator): def factorial(self): return UnaryExpression(self.expr, modifier=operators.custom_op("!"), type_=MyInteger) def factorial_prefix(self): return UnaryExpression(self.expr, operator=operators.custom_op("!!"), type_=MyInteger) def __invert__(self): return UnaryExpression(self.expr, operator=operators.custom_op("!!!"), type_=MyInteger) return MyInteger def test_factorial(self): col = column('somecol', self._factorial_fixture()) self.assert_compile( col.factorial(), "somecol !" ) def test_double_factorial(self): col = column('somecol', self._factorial_fixture()) self.assert_compile( col.factorial().factorial(), "somecol ! !" ) def test_factorial_prefix(self): col = column('somecol', self._factorial_fixture()) self.assert_compile( col.factorial_prefix(), "!! somecol" ) def test_factorial_invert(self): col = column('somecol', self._factorial_fixture()) self.assert_compile( ~col, "!!! somecol" ) def test_double_factorial_invert(self): col = column('somecol', self._factorial_fixture()) self.assert_compile( ~(~col), "!!! (!!! somecol)" ) def test_unary_no_ops(self): assert_raises_message( exc.CompileError, "Unary expression has no operator or modifier", UnaryExpression(literal("x")).compile ) def test_unary_both_ops(self): assert_raises_message( exc.CompileError, "Unary expression does not support operator and " "modifier simultaneously", UnaryExpression(literal("x"), operator=operators.custom_op("x"), modifier=operators.custom_op("y")).compile ) class _CustomComparatorTests(object): def test_override_builtin(self): c1 = Column('foo', self._add_override_factory()) self._assert_add_override(c1) def test_column_proxy(self): t = Table('t', MetaData(), Column('foo', self._add_override_factory()) ) proxied = t.select().c.foo self._assert_add_override(proxied) self._assert_and_override(proxied) def test_alias_proxy(self): t = Table('t', MetaData(), Column('foo', self._add_override_factory()) ) proxied = t.alias().c.foo self._assert_add_override(proxied) self._assert_and_override(proxied) def test_binary_propagate(self): c1 = Column('foo', self._add_override_factory()) self._assert_add_override(c1 - 6) self._assert_and_override(c1 - 6) def test_reverse_binary_propagate(self): c1 = Column('foo', self._add_override_factory()) self._assert_add_override(6 - c1) self._assert_and_override(6 - c1) def test_binary_multi_propagate(self): c1 = Column('foo', self._add_override_factory()) self._assert_add_override((c1 - 6) + 5) self._assert_and_override((c1 - 6) + 5) def test_no_boolean_propagate(self): c1 = Column('foo', self._add_override_factory()) self._assert_not_add_override(c1 == 56) self._assert_not_and_override(c1 == 56) def _assert_and_override(self, expr): assert (expr & text("5")).compare( expr.op("goofy_and")(text("5")) ) def _assert_add_override(self, expr): assert (expr + 5).compare( expr.op("goofy")(5) ) def _assert_not_add_override(self, expr): assert not (expr + 5).compare( expr.op("goofy")(5) ) def _assert_not_and_override(self, expr): assert not (expr & text("5")).compare( expr.op("goofy_and")(text("5")) ) class CustomComparatorTest(_CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class MyInteger(Integer): class comparator_factory(TypeEngine.Comparator): def __init__(self, expr): super(MyInteger.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("goofy")(other) def __and__(self, other): return self.expr.op("goofy_and")(other) return MyInteger class TypeDecoratorComparatorTest(_CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class MyInteger(TypeDecorator): impl = Integer class comparator_factory(TypeDecorator.Comparator): def __init__(self, expr): super(MyInteger.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("goofy")(other) def __and__(self, other): return self.expr.op("goofy_and")(other) return MyInteger class TypeDecoratorTypeDecoratorComparatorTest( _CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class MyIntegerOne(TypeDecorator): impl = Integer class comparator_factory(TypeDecorator.Comparator): def __init__(self, expr): super(MyIntegerOne.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("goofy")(other) def __and__(self, other): return self.expr.op("goofy_and")(other) class MyIntegerTwo(TypeDecorator): impl = MyIntegerOne return MyIntegerTwo class TypeDecoratorWVariantComparatorTest( _CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class SomeOtherInteger(Integer): class comparator_factory(TypeEngine.Comparator): def __init__(self, expr): super( SomeOtherInteger.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("not goofy")(other) def __and__(self, other): return self.expr.op("not goofy_and")(other) class MyInteger(TypeDecorator): impl = Integer class comparator_factory(TypeDecorator.Comparator): def __init__(self, expr): super(MyInteger.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("goofy")(other) def __and__(self, other): return self.expr.op("goofy_and")(other) return MyInteger().with_variant(SomeOtherInteger, "mysql") class CustomEmbeddedinTypeDecoratorTest( _CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class MyInteger(Integer): class comparator_factory(TypeEngine.Comparator): def __init__(self, expr): super(MyInteger.comparator_factory, self).__init__(expr) def __add__(self, other): return self.expr.op("goofy")(other) def __and__(self, other): return self.expr.op("goofy_and")(other) class MyDecInteger(TypeDecorator): impl = MyInteger return MyDecInteger class NewOperatorTest(_CustomComparatorTests, fixtures.TestBase): def _add_override_factory(self): class MyInteger(Integer): class comparator_factory(TypeEngine.Comparator): def __init__(self, expr): super(MyInteger.comparator_factory, self).__init__(expr) def foob(self, other): return self.expr.op("foob")(other) return MyInteger def _assert_add_override(self, expr): assert (expr.foob(5)).compare( expr.op("foob")(5) ) def _assert_not_add_override(self, expr): assert not hasattr(expr, "foob") def _assert_and_override(self, expr): pass def _assert_not_and_override(self, expr): pass class ExtensionOperatorTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' def test_contains(self): class MyType(UserDefinedType): class comparator_factory(UserDefinedType.Comparator): def contains(self, other, **kw): return self.op("->")(other) self.assert_compile( Column('x', MyType()).contains(5), "x -> :x_1" ) def test_getitem(self): class MyType(UserDefinedType): class comparator_factory(UserDefinedType.Comparator): def __getitem__(self, index): return self.op("->")(index) self.assert_compile( Column('x', MyType())[5], "x -> :x_1" ) def test_op_not_an_iterator(self): # see [ticket:2726] class MyType(UserDefinedType): class comparator_factory(UserDefinedType.Comparator): def __getitem__(self, index): return self.op("->")(index) col = Column('x', MyType()) assert not isinstance(col, collections.Iterable) def test_lshift(self): class MyType(UserDefinedType): class comparator_factory(UserDefinedType.Comparator): def __lshift__(self, other): return self.op("->")(other) self.assert_compile( Column('x', MyType()) << 5, "x -> :x_1" ) def test_rshift(self): class MyType(UserDefinedType): class comparator_factory(UserDefinedType.Comparator): def __rshift__(self, other): return self.op("->")(other) self.assert_compile( Column('x', MyType()) >> 5, "x -> :x_1" ) class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL): """test standalone booleans being wrapped in an AsBoolean, as well as true/false compilation.""" def _dialect(self, native_boolean): d = default.DefaultDialect() d.supports_native_boolean = native_boolean return d def test_one(self): c = column('x', Boolean) self.assert_compile( select([c]).where(c), "SELECT x WHERE x", dialect=self._dialect(True) ) def test_two_a(self): c = column('x', Boolean) self.assert_compile( select([c]).where(c), "SELECT x WHERE x = 1", dialect=self._dialect(False) ) def test_two_b(self): c = column('x', Boolean) self.assert_compile( select([c], whereclause=c), "SELECT x WHERE x = 1", dialect=self._dialect(False) ) def test_three_a(self): c = column('x', Boolean) self.assert_compile( select([c]).where(~c), "SELECT x WHERE x = 0", dialect=self._dialect(False) ) def test_three_b(self): c = column('x', Boolean) self.assert_compile( select([c], whereclause=~c), "SELECT x WHERE x = 0", dialect=self._dialect(False) ) def test_four(self): c = column('x', Boolean) self.assert_compile( select([c]).where(~c), "SELECT x WHERE NOT x", dialect=self._dialect(True) ) def test_five_a(self): c = column('x', Boolean) self.assert_compile( select([c]).having(c), "SELECT x HAVING x = 1", dialect=self._dialect(False) ) def test_five_b(self): c = column('x', Boolean) self.assert_compile( select([c], having=c), "SELECT x HAVING x = 1", dialect=self._dialect(False) ) def test_six(self): self.assert_compile( or_(false(), true()), "1 = 1", dialect=self._dialect(False) ) def test_eight(self): self.assert_compile( and_(false(), true()), "false", dialect=self._dialect(True) ) def test_nine(self): self.assert_compile( and_(false(), true()), "0 = 1", dialect=self._dialect(False) ) def test_ten(self): c = column('x', Boolean) self.assert_compile( c == 1, "x = :x_1", dialect=self._dialect(False) ) def test_eleven(self): c = column('x', Boolean) self.assert_compile( c.is_(true()), "x IS true", dialect=self._dialect(True) ) def test_twelve(self): c = column('x', Boolean) # I don't have a solution for this one yet, # other than adding some heavy-handed conditionals # into compiler self.assert_compile( c.is_(true()), "x IS 1", dialect=self._dialect(False) ) class ConjunctionTest(fixtures.TestBase, testing.AssertsCompiledSQL): """test interaction of and_()/or_() with boolean , null constants """ __dialect__ = default.DefaultDialect(supports_native_boolean=True) def test_one(self): self.assert_compile(~and_(true()), "false") def test_two(self): self.assert_compile(or_(~and_(true())), "false") def test_three(self): self.assert_compile(or_(and_()), "") def test_four(self): x = column('x') self.assert_compile( and_(or_(x == 5), or_(x == 7)), "x = :x_1 AND x = :x_2") def test_five(self): x = column("x") self.assert_compile( and_(true()._ifnone(None), x == 7), "x = :x_1" ) def test_six(self): x = column("x") self.assert_compile(or_(true(), x == 7), "true") self.assert_compile(or_(x == 7, true()), "true") self.assert_compile(~or_(x == 7, true()), "false") def test_six_pt_five(self): x = column("x") self.assert_compile(select([x]).where(or_(x == 7, true())), "SELECT x WHERE true") self.assert_compile( select( [x]).where( or_( x == 7, true())), "SELECT x WHERE 1 = 1", dialect=default.DefaultDialect( supports_native_boolean=False)) def test_seven(self): x = column("x") self.assert_compile( and_(true(), x == 7, true(), x == 9), "x = :x_1 AND x = :x_2") def test_eight(self): x = column("x") self.assert_compile( or_(false(), x == 7, false(), x == 9), "x = :x_1 OR x = :x_2") def test_nine(self): x = column("x") self.assert_compile( and_(x == 7, x == 9, false(), x == 5), "false" ) self.assert_compile( ~and_(x == 7, x == 9, false(), x == 5), "true" ) def test_ten(self): self.assert_compile( and_(None, None), "NULL AND NULL" ) def test_eleven(self): x = column("x") self.assert_compile( select([x]).where(None).where(None), "SELECT x WHERE NULL AND NULL" ) def test_twelve(self): x = column("x") self.assert_compile( select([x]).where(and_(None, None)), "SELECT x WHERE NULL AND NULL" ) def test_thirteen(self): x = column("x") self.assert_compile( select([x]).where(~and_(None, None)), "SELECT x WHERE NOT (NULL AND NULL)" ) def test_fourteen(self): x = column("x") self.assert_compile( select([x]).where(~null()), "SELECT x WHERE NOT NULL" ) def test_constant_non_singleton(self): is_not_(null(), null()) is_not_(false(), false()) is_not_(true(), true()) def test_constant_render_distinct(self): self.assert_compile( select([null(), null()]), "SELECT NULL AS anon_1, NULL AS anon_2" ) self.assert_compile( select([true(), true()]), "SELECT true AS anon_1, true AS anon_2" ) self.assert_compile( select([false(), false()]), "SELECT false AS anon_1, false AS anon_2" ) class OperatorPrecedenceTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), column('name', String), column('description', String), ) table2 = table('op', column('field')) def test_operator_precedence_1(self): self.assert_compile( self.table2.select((self.table2.c.field == 5) == None), "SELECT op.field FROM op WHERE (op.field = :field_1) IS NULL") def test_operator_precedence_2(self): self.assert_compile( self.table2.select( (self.table2.c.field + 5) == self.table2.c.field), "SELECT op.field FROM op WHERE op.field + :field_1 = op.field") def test_operator_precedence_3(self): self.assert_compile( self.table2.select((self.table2.c.field + 5) * 6), "SELECT op.field FROM op WHERE (op.field + :field_1) * :param_1") def test_operator_precedence_4(self): self.assert_compile( self.table2.select( (self.table2.c.field * 5) + 6), "SELECT op.field FROM op WHERE op.field * :field_1 + :param_1") def test_operator_precedence_5(self): self.assert_compile(self.table2.select( 5 + self.table2.c.field.in_([5, 6])), "SELECT op.field FROM op WHERE :param_1 + " "(op.field IN (:field_1, :field_2))") def test_operator_precedence_6(self): self.assert_compile(self.table2.select( (5 + self.table2.c.field).in_([5, 6])), "SELECT op.field FROM op WHERE :field_1 + op.field " "IN (:param_1, :param_2)") def test_operator_precedence_7(self): self.assert_compile(self.table2.select( not_(and_(self.table2.c.field == 5, self.table2.c.field == 7))), "SELECT op.field FROM op WHERE NOT " "(op.field = :field_1 AND op.field = :field_2)") def test_operator_precedence_8(self): self.assert_compile( self.table2.select( not_( self.table2.c.field == 5)), "SELECT op.field FROM op WHERE op.field != :field_1") def test_operator_precedence_9(self): self.assert_compile(self.table2.select( not_(self.table2.c.field.between(5, 6))), "SELECT op.field FROM op WHERE " "op.field NOT BETWEEN :field_1 AND :field_2") def test_operator_precedence_10(self): self.assert_compile( self.table2.select( not_( self.table2.c.field) == 5), "SELECT op.field FROM op WHERE (NOT op.field) = :param_1") def test_operator_precedence_11(self): self.assert_compile(self.table2.select( (self.table2.c.field == self.table2.c.field). between(False, True)), "SELECT op.field FROM op WHERE (op.field = op.field) " "BETWEEN :param_1 AND :param_2") def test_operator_precedence_12(self): self.assert_compile(self.table2.select( between((self.table2.c.field == self.table2.c.field), False, True)), "SELECT op.field FROM op WHERE (op.field = op.field) " "BETWEEN :param_1 AND :param_2") def test_operator_precedence_13(self): self.assert_compile( self.table2.select( self.table2.c.field.match( self.table2.c.field).is_(None)), "SELECT op.field FROM op WHERE (op.field MATCH op.field) IS NULL") def test_operator_precedence_collate_1(self): self.assert_compile( self.table1.c.name == literal('foo').collate('utf-8'), "mytable.name = (:param_1 COLLATE utf-8)" ) def test_operator_precedence_collate_2(self): self.assert_compile( (self.table1.c.name == literal('foo')).collate('utf-8'), "mytable.name = :param_1 COLLATE utf-8" ) def test_operator_precedence_collate_3(self): self.assert_compile( self.table1.c.name.collate('utf-8') == 'foo', "(mytable.name COLLATE utf-8) = :param_1" ) def test_operator_precedence_collate_4(self): self.assert_compile( and_( (self.table1.c.name == literal('foo')).collate('utf-8'), (self.table2.c.field == literal('bar')).collate('utf-8'), ), "mytable.name = :param_1 COLLATE utf-8 " "AND op.field = :param_2 COLLATE utf-8" ) def test_operator_precedence_collate_5(self): self.assert_compile( select([self.table1.c.name]).order_by( self.table1.c.name.collate('utf-8').desc()), "SELECT mytable.name FROM mytable " "ORDER BY mytable.name COLLATE utf-8 DESC" ) def test_operator_precedence_collate_6(self): self.assert_compile( select([self.table1.c.name]).order_by( self.table1.c.name.collate('utf-8').desc().nullslast()), "SELECT mytable.name FROM mytable " "ORDER BY mytable.name COLLATE utf-8 DESC NULLS LAST" ) def test_operator_precedence_collate_7(self): self.assert_compile( select([self.table1.c.name]).order_by( self.table1.c.name.collate('utf-8').asc()), "SELECT mytable.name FROM mytable " "ORDER BY mytable.name COLLATE utf-8 ASC" ) def test_commutative_operators(self): self.assert_compile( literal("a") + literal("b") * literal("c"), ":param_1 || :param_2 * :param_3" ) def test_op_operators(self): self.assert_compile( self.table1.select(self.table1.c.myid.op('hoho')(12) == 14), "SELECT mytable.myid, mytable.name, mytable.description FROM " "mytable WHERE (mytable.myid hoho :myid_1) = :param_1" ) def test_op_operators_comma_precedence(self): self.assert_compile( func.foo(self.table1.c.myid.op('hoho')(12)), "foo(mytable.myid hoho :myid_1)" ) def test_op_operators_comparison_precedence(self): self.assert_compile( self.table1.c.myid.op('hoho')(12) == 5, "(mytable.myid hoho :myid_1) = :param_1" ) def test_op_operators_custom_precedence(self): op1 = self.table1.c.myid.op('hoho', precedence=5) op2 = op1(5).op('lala', precedence=4)(4) op3 = op1(5).op('lala', precedence=6)(4) self.assert_compile(op2, "mytable.myid hoho :myid_1 lala :param_1") self.assert_compile(op3, "(mytable.myid hoho :myid_1) lala :param_1") class OperatorAssociativityTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' def test_associativity_1(self): f = column('f') self.assert_compile(f - f, "f - f") def test_associativity_2(self): f = column('f') self.assert_compile(f - f - f, "(f - f) - f") def test_associativity_3(self): f = column('f') self.assert_compile((f - f) - f, "(f - f) - f") def test_associativity_4(self): f = column('f') self.assert_compile((f - f).label('foo') - f, "(f - f) - f") def test_associativity_5(self): f = column('f') self.assert_compile(f - (f - f), "f - (f - f)") def test_associativity_6(self): f = column('f') self.assert_compile(f - (f - f).label('foo'), "f - (f - f)") def test_associativity_7(self): f = column('f') # because - less precedent than / self.assert_compile(f / (f - f), "f / (f - f)") def test_associativity_8(self): f = column('f') self.assert_compile(f / (f - f).label('foo'), "f / (f - f)") def test_associativity_9(self): f = column('f') self.assert_compile(f / f - f, "f / f - f") def test_associativity_10(self): f = column('f') self.assert_compile((f / f) - f, "f / f - f") def test_associativity_11(self): f = column('f') self.assert_compile((f / f).label('foo') - f, "f / f - f") def test_associativity_12(self): f = column('f') # because / more precedent than - self.assert_compile(f - (f / f), "f - f / f") def test_associativity_13(self): f = column('f') self.assert_compile(f - (f / f).label('foo'), "f - f / f") def test_associativity_14(self): f = column('f') self.assert_compile(f - f / f, "f - f / f") def test_associativity_15(self): f = column('f') self.assert_compile((f - f) / f, "(f - f) / f") def test_associativity_16(self): f = column('f') self.assert_compile(((f - f) / f) - f, "(f - f) / f - f") def test_associativity_17(self): f = column('f') # - lower precedence than / self.assert_compile((f - f) / (f - f), "(f - f) / (f - f)") def test_associativity_18(self): f = column('f') # / higher precedence than - self.assert_compile((f / f) - (f / f), "f / f - f / f") def test_associativity_19(self): f = column('f') self.assert_compile((f / f) - (f - f), "f / f - (f - f)") def test_associativity_20(self): f = column('f') self.assert_compile((f / f) / (f - f), "(f / f) / (f - f)") def test_associativity_21(self): f = column('f') self.assert_compile(f / (f / (f - f)), "f / (f / (f - f))") class InTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), ) table2 = table( 'myothertable', column('otherid', Integer), column('othername', String) ) def test_in_1(self): self.assert_compile(self.table1.c.myid.in_(['a']), "mytable.myid IN (:myid_1)") def test_in_2(self): self.assert_compile(~self.table1.c.myid.in_(['a']), "mytable.myid NOT IN (:myid_1)") def test_in_3(self): self.assert_compile(self.table1.c.myid.in_(['a', 'b']), "mytable.myid IN (:myid_1, :myid_2)") def test_in_4(self): self.assert_compile(self.table1.c.myid.in_(iter(['a', 'b'])), "mytable.myid IN (:myid_1, :myid_2)") def test_in_5(self): self.assert_compile(self.table1.c.myid.in_([literal('a')]), "mytable.myid IN (:param_1)") def test_in_6(self): self.assert_compile(self.table1.c.myid.in_([literal('a'), 'b']), "mytable.myid IN (:param_1, :myid_1)") def test_in_7(self): self.assert_compile( self.table1.c.myid.in_([literal('a'), literal('b')]), "mytable.myid IN (:param_1, :param_2)") def test_in_8(self): self.assert_compile(self.table1.c.myid.in_(['a', literal('b')]), "mytable.myid IN (:myid_1, :param_1)") def test_in_9(self): self.assert_compile(self.table1.c.myid.in_([literal(1) + 'a']), "mytable.myid IN (:param_1 + :param_2)") def test_in_10(self): self.assert_compile(self.table1.c.myid.in_([literal('a') + 'a', 'b']), "mytable.myid IN (:param_1 || :param_2, :myid_1)") def test_in_11(self): self.assert_compile( self.table1.c.myid.in_( [ literal('a') + literal('a'), literal('b')]), "mytable.myid IN (:param_1 || :param_2, :param_3)") def test_in_12(self): self.assert_compile(self.table1.c.myid.in_([1, literal(3) + 4]), "mytable.myid IN (:myid_1, :param_1 + :param_2)") def test_in_13(self): self.assert_compile(self.table1.c.myid.in_([literal('a') < 'b']), "mytable.myid IN (:param_1 < :param_2)") def test_in_14(self): self.assert_compile(self.table1.c.myid.in_([self.table1.c.myid]), "mytable.myid IN (mytable.myid)") def test_in_15(self): self.assert_compile(self.table1.c.myid.in_(['a', self.table1.c.myid]), "mytable.myid IN (:myid_1, mytable.myid)") def test_in_16(self): self.assert_compile(self.table1.c.myid.in_([literal('a'), self.table1.c.myid]), "mytable.myid IN (:param_1, mytable.myid)") def test_in_17(self): self.assert_compile( self.table1.c.myid.in_( [ literal('a'), self.table1.c.myid + 'a']), "mytable.myid IN (:param_1, mytable.myid + :myid_1)") def test_in_18(self): self.assert_compile( self.table1.c.myid.in_( [ literal(1), 'a' + self.table1.c.myid]), "mytable.myid IN (:param_1, :myid_1 + mytable.myid)") def test_in_19(self): self.assert_compile(self.table1.c.myid.in_([1, 2, 3]), "mytable.myid IN (:myid_1, :myid_2, :myid_3)") def test_in_20(self): self.assert_compile(self.table1.c.myid.in_( select([self.table2.c.otherid])), "mytable.myid IN (SELECT myothertable.otherid FROM myothertable)") def test_in_21(self): self.assert_compile(~self.table1.c.myid.in_( select([self.table2.c.otherid])), "mytable.myid NOT IN (SELECT myothertable.otherid FROM myothertable)") def test_in_22(self): self.assert_compile( self.table1.c.myid.in_( text("SELECT myothertable.otherid FROM myothertable") ), "mytable.myid IN (SELECT myothertable.otherid " "FROM myothertable)" ) @testing.emits_warning('.*empty sequence.*') def test_in_23(self): self.assert_compile(self.table1.c.myid.in_([]), "mytable.myid != mytable.myid") def test_in_24(self): self.assert_compile( select([self.table1.c.myid.in_(select([self.table2.c.otherid]))]), "SELECT mytable.myid IN (SELECT myothertable.otherid " "FROM myothertable) AS anon_1 FROM mytable" ) def test_in_25(self): self.assert_compile( select([self.table1.c.myid.in_( select([self.table2.c.otherid]).as_scalar())]), "SELECT mytable.myid IN (SELECT myothertable.otherid " "FROM myothertable) AS anon_1 FROM mytable" ) def test_in_26(self): self.assert_compile(self.table1.c.myid.in_( union( select([self.table1.c.myid], self.table1.c.myid == 5), select([self.table1.c.myid], self.table1.c.myid == 12), ) ), "mytable.myid IN (" "SELECT mytable.myid FROM mytable WHERE mytable.myid = :myid_1 " "UNION SELECT mytable.myid FROM mytable WHERE mytable.myid = :myid_2)") def test_in_27(self): # test that putting a select in an IN clause does not # blow away its ORDER BY clause self.assert_compile( select([self.table1, self.table2], self.table2.c.otherid.in_( select([self.table2.c.otherid], order_by=[self.table2.c.othername], limit=10, correlate=False) ), from_obj=[self.table1.join(self.table2, self.table1.c.myid == self.table2.c.otherid)], order_by=[self.table1.c.myid] ), "SELECT mytable.myid, " "myothertable.otherid, myothertable.othername FROM mytable " "JOIN myothertable ON mytable.myid = myothertable.otherid " "WHERE myothertable.otherid IN (SELECT myothertable.otherid " "FROM myothertable ORDER BY myothertable.othername " "LIMIT :param_1) ORDER BY mytable.myid", {'param_1': 10} ) def test_in_28(self): self.assert_compile( self.table1.c.myid.in_([None]), "mytable.myid IN (NULL)" ) @testing.emits_warning('.*empty sequence.*') def test_in_29(self): self.assert_compile(self.table1.c.myid.notin_([]), "mytable.myid = mytable.myid") @testing.emits_warning('.*empty sequence.*') def test_in_30(self): self.assert_compile(~self.table1.c.myid.in_([]), "mytable.myid = mytable.myid") class MathOperatorTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), ) def _test_math_op(self, py_op, sql_op): for (lhs, rhs, res) in ( (5, self.table1.c.myid, ':myid_1 %s mytable.myid'), (5, literal(5), ':param_1 %s :param_2'), (self.table1.c.myid, 'b', 'mytable.myid %s :myid_1'), (self.table1.c.myid, literal(2.7), 'mytable.myid %s :param_1'), (self.table1.c.myid, self.table1.c.myid, 'mytable.myid %s mytable.myid'), (literal(5), 8, ':param_1 %s :param_2'), (literal(6), self.table1.c.myid, ':param_1 %s mytable.myid'), (literal(7), literal(5.5), ':param_1 %s :param_2'), ): self.assert_compile(py_op(lhs, rhs), res % sql_op) def test_math_op_add(self): self._test_math_op(operator.add, '+') def test_math_op_mul(self): self._test_math_op(operator.mul, '*') def test_math_op_sub(self): self._test_math_op(operator.sub, '-') def test_math_op_div(self): if util.py3k: self._test_math_op(operator.truediv, '/') else: self._test_math_op(operator.div, '/') class ComparisonOperatorTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), ) def test_pickle_operators_one(self): clause = (self.table1.c.myid == 12) & \ self.table1.c.myid.between(15, 20) & \ self.table1.c.myid.like('hoho') eq_(str(clause), str(util.pickle.loads(util.pickle.dumps(clause)))) def test_pickle_operators_two(self): clause = tuple_(1, 2, 3) eq_(str(clause), str(util.pickle.loads(util.pickle.dumps(clause)))) def _test_comparison_op(self, py_op, fwd_op, rev_op): dt = datetime.datetime(2012, 5, 10, 15, 27, 18) for (lhs, rhs, l_sql, r_sql) in ( ('a', self.table1.c.myid, ':myid_1', 'mytable.myid'), ('a', literal('b'), ':param_2', ':param_1'), # note swap! (self.table1.c.myid, 'b', 'mytable.myid', ':myid_1'), (self.table1.c.myid, literal('b'), 'mytable.myid', ':param_1'), (self.table1.c.myid, self.table1.c.myid, 'mytable.myid', 'mytable.myid'), (literal('a'), 'b', ':param_1', ':param_2'), (literal('a'), self.table1.c.myid, ':param_1', 'mytable.myid'), (literal('a'), literal('b'), ':param_1', ':param_2'), (dt, literal('b'), ':param_2', ':param_1'), (literal('b'), dt, ':param_1', ':param_2'), ): # the compiled clause should match either (e.g.): # 'a' < 'b' -or- 'b' > 'a'. compiled = str(py_op(lhs, rhs)) fwd_sql = "%s %s %s" % (l_sql, fwd_op, r_sql) rev_sql = "%s %s %s" % (r_sql, rev_op, l_sql) self.assert_(compiled == fwd_sql or compiled == rev_sql, "\n'" + compiled + "'\n does not match\n'" + fwd_sql + "'\n or\n'" + rev_sql + "'") def test_comparison_operators_lt(self): self._test_comparison_op(operator.lt, '<', '>'), def test_comparison_operators_gt(self): self._test_comparison_op(operator.gt, '>', '<') def test_comparison_operators_eq(self): self._test_comparison_op(operator.eq, '=', '=') def test_comparison_operators_ne(self): self._test_comparison_op(operator.ne, '!=', '!=') def test_comparison_operators_le(self): self._test_comparison_op(operator.le, '<=', '>=') def test_comparison_operators_ge(self): self._test_comparison_op(operator.ge, '>=', '<=') class NonZeroTest(fixtures.TestBase): def _raises(self, expr): assert_raises_message( TypeError, "Boolean value of this clause is not defined", bool, expr ) def _assert_true(self, expr): is_(bool(expr), True) def _assert_false(self, expr): is_(bool(expr), False) def test_column_identity_eq(self): c1 = column('c1') self._assert_true(c1 == c1) def test_column_identity_gt(self): c1 = column('c1') self._raises(c1 > c1) def test_column_compare_eq(self): c1, c2 = column('c1'), column('c2') self._assert_false(c1 == c2) def test_column_compare_gt(self): c1, c2 = column('c1'), column('c2') self._raises(c1 > c2) def test_binary_identity_eq(self): c1 = column('c1') expr = c1 > 5 self._assert_true(expr == expr) def test_labeled_binary_identity_eq(self): c1 = column('c1') expr = (c1 > 5).label(None) self._assert_true(expr == expr) def test_annotated_binary_identity_eq(self): c1 = column('c1') expr1 = (c1 > 5) expr2 = expr1._annotate({"foo": "bar"}) self._assert_true(expr1 == expr2) def test_labeled_binary_compare_gt(self): c1 = column('c1') expr1 = (c1 > 5).label(None) expr2 = (c1 > 5).label(None) self._assert_false(expr1 == expr2) class NegationTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), column('name', String), ) def test_negate_operators_1(self): for (py_op, op) in ( (operator.neg, '-'), (operator.inv, 'NOT '), ): for expr, expected in ( (self.table1.c.myid, "mytable.myid"), (literal("foo"), ":param_1"), ): self.assert_compile(py_op(expr), "%s%s" % (op, expected)) def test_negate_operators_2(self): self.assert_compile( self.table1.select((self.table1.c.myid != 12) & ~(self.table1.c.name == 'john')), "SELECT mytable.myid, mytable.name FROM " "mytable WHERE mytable.myid != :myid_1 " "AND mytable.name != :name_1" ) def test_negate_operators_3(self): self.assert_compile( self.table1.select((self.table1.c.myid != 12) & ~(self.table1.c.name.between('jack', 'john'))), "SELECT mytable.myid, mytable.name FROM " "mytable WHERE mytable.myid != :myid_1 AND " "mytable.name NOT BETWEEN :name_1 AND :name_2" ) def test_negate_operators_4(self): self.assert_compile( self.table1.select((self.table1.c.myid != 12) & ~and_(self.table1.c.name == 'john', self.table1.c.name == 'ed', self.table1.c.name == 'fred')), "SELECT mytable.myid, mytable.name FROM " "mytable WHERE mytable.myid != :myid_1 AND " "NOT (mytable.name = :name_1 AND mytable.name = :name_2 " "AND mytable.name = :name_3)" ) def test_negate_operators_5(self): self.assert_compile( self.table1.select( (self.table1.c.myid != 12) & ~self.table1.c.name), "SELECT mytable.myid, mytable.name FROM " "mytable WHERE mytable.myid != :myid_1 AND NOT mytable.name") class LikeTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), column('name', String), ) def test_like_1(self): self.assert_compile( self.table1.c.myid.like('somstr'), "mytable.myid LIKE :myid_1") def test_like_2(self): self.assert_compile( ~self.table1.c.myid.like('somstr'), "mytable.myid NOT LIKE :myid_1") def test_like_3(self): self.assert_compile( self.table1.c.myid.like('somstr', escape='\\'), "mytable.myid LIKE :myid_1 ESCAPE '\\'") def test_like_4(self): self.assert_compile( ~self.table1.c.myid.like('somstr', escape='\\'), "mytable.myid NOT LIKE :myid_1 ESCAPE '\\'") def test_like_5(self): self.assert_compile( self.table1.c.myid.ilike('somstr', escape='\\'), "lower(mytable.myid) LIKE lower(:myid_1) ESCAPE '\\'") def test_like_6(self): self.assert_compile( ~self.table1.c.myid.ilike('somstr', escape='\\'), "lower(mytable.myid) NOT LIKE lower(:myid_1) ESCAPE '\\'") def test_like_7(self): self.assert_compile( self.table1.c.myid.ilike('somstr', escape='\\'), "mytable.myid ILIKE %(myid_1)s ESCAPE '\\\\'", dialect=postgresql.dialect()) def test_like_8(self): self.assert_compile( ~self.table1.c.myid.ilike('somstr', escape='\\'), "mytable.myid NOT ILIKE %(myid_1)s ESCAPE '\\\\'", dialect=postgresql.dialect()) def test_like_9(self): self.assert_compile( self.table1.c.name.ilike('%something%'), "lower(mytable.name) LIKE lower(:name_1)") def test_like_10(self): self.assert_compile( self.table1.c.name.ilike('%something%'), "mytable.name ILIKE %(name_1)s", dialect=postgresql.dialect()) def test_like_11(self): self.assert_compile( ~self.table1.c.name.ilike('%something%'), "lower(mytable.name) NOT LIKE lower(:name_1)") def test_like_12(self): self.assert_compile( ~self.table1.c.name.ilike('%something%'), "mytable.name NOT ILIKE %(name_1)s", dialect=postgresql.dialect()) class BetweenTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), column('name', String), ) def test_between_1(self): self.assert_compile( self.table1.c.myid.between(1, 2), "mytable.myid BETWEEN :myid_1 AND :myid_2") def test_between_2(self): self.assert_compile( ~self.table1.c.myid.between(1, 2), "mytable.myid NOT BETWEEN :myid_1 AND :myid_2") def test_between_3(self): self.assert_compile( self.table1.c.myid.between(1, 2, symmetric=True), "mytable.myid BETWEEN SYMMETRIC :myid_1 AND :myid_2") def test_between_4(self): self.assert_compile( ~self.table1.c.myid.between(1, 2, symmetric=True), "mytable.myid NOT BETWEEN SYMMETRIC :myid_1 AND :myid_2") def test_between_5(self): self.assert_compile( between(self.table1.c.myid, 1, 2, symmetric=True), "mytable.myid BETWEEN SYMMETRIC :myid_1 AND :myid_2") def test_between_6(self): self.assert_compile( ~between(self.table1.c.myid, 1, 2, symmetric=True), "mytable.myid NOT BETWEEN SYMMETRIC :myid_1 AND :myid_2") class MatchTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' table1 = table('mytable', column('myid', Integer), column('name', String), ) def test_match_1(self): self.assert_compile(self.table1.c.myid.match('somstr'), "mytable.myid MATCH ?", dialect=sqlite.dialect()) def test_match_2(self): self.assert_compile( self.table1.c.myid.match('somstr'), "MATCH (mytable.myid) AGAINST (%s IN BOOLEAN MODE)", dialect=mysql.dialect()) def test_match_3(self): self.assert_compile(self.table1.c.myid.match('somstr'), "CONTAINS (mytable.myid, :myid_1)", dialect=mssql.dialect()) def test_match_4(self): self.assert_compile(self.table1.c.myid.match('somstr'), "mytable.myid @@ to_tsquery(%(myid_1)s)", dialect=postgresql.dialect()) def test_match_5(self): self.assert_compile(self.table1.c.myid.match('somstr'), "CONTAINS (mytable.myid, :myid_1)", dialect=oracle.dialect()) def test_match_is_now_matchtype(self): expr = self.table1.c.myid.match('somstr') assert expr.type._type_affinity is MatchType()._type_affinity assert isinstance(expr.type, MatchType) def test_boolean_inversion_postgresql(self): self.assert_compile( ~self.table1.c.myid.match('somstr'), "NOT mytable.myid @@ to_tsquery(%(myid_1)s)", dialect=postgresql.dialect()) def test_boolean_inversion_mysql(self): # because mysql doesnt have native boolean self.assert_compile( ~self.table1.c.myid.match('somstr'), "NOT MATCH (mytable.myid) AGAINST (%s IN BOOLEAN MODE)", dialect=mysql.dialect()) def test_boolean_inversion_mssql(self): # because mssql doesnt have native boolean self.assert_compile( ~self.table1.c.myid.match('somstr'), "NOT CONTAINS (mytable.myid, :myid_1)", dialect=mssql.dialect()) class ComposedLikeOperatorsTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = 'default' def test_contains(self): self.assert_compile( column('x').contains('y'), "x LIKE '%%' || :x_1 || '%%'", checkparams={'x_1': 'y'} ) def test_contains_escape(self): self.assert_compile( column('x').contains('y', escape='\\'), "x LIKE '%%' || :x_1 || '%%' ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_contains_literal(self): self.assert_compile( column('x').contains(literal_column('y')), "x LIKE '%%' || y || '%%'", checkparams={} ) def test_contains_text(self): self.assert_compile( column('x').contains(text('y')), "x LIKE '%%' || y || '%%'", checkparams={} ) def test_not_contains(self): self.assert_compile( ~column('x').contains('y'), "x NOT LIKE '%%' || :x_1 || '%%'", checkparams={'x_1': 'y'} ) def test_not_contains_escape(self): self.assert_compile( ~column('x').contains('y', escape='\\'), "x NOT LIKE '%%' || :x_1 || '%%' ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_contains_concat(self): self.assert_compile( column('x').contains('y'), "x LIKE concat(concat('%%', %s), '%%')", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_not_contains_concat(self): self.assert_compile( ~column('x').contains('y'), "x NOT LIKE concat(concat('%%', %s), '%%')", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_contains_literal_concat(self): self.assert_compile( column('x').contains(literal_column('y')), "x LIKE concat(concat('%%', y), '%%')", checkparams={}, dialect=mysql.dialect() ) def test_contains_text_concat(self): self.assert_compile( column('x').contains(text('y')), "x LIKE concat(concat('%%', y), '%%')", checkparams={}, dialect=mysql.dialect() ) def test_startswith(self): self.assert_compile( column('x').startswith('y'), "x LIKE :x_1 || '%%'", checkparams={'x_1': 'y'} ) def test_startswith_escape(self): self.assert_compile( column('x').startswith('y', escape='\\'), "x LIKE :x_1 || '%%' ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_not_startswith(self): self.assert_compile( ~column('x').startswith('y'), "x NOT LIKE :x_1 || '%%'", checkparams={'x_1': 'y'} ) def test_not_startswith_escape(self): self.assert_compile( ~column('x').startswith('y', escape='\\'), "x NOT LIKE :x_1 || '%%' ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_startswith_literal(self): self.assert_compile( column('x').startswith(literal_column('y')), "x LIKE y || '%%'", checkparams={} ) def test_startswith_text(self): self.assert_compile( column('x').startswith(text('y')), "x LIKE y || '%%'", checkparams={} ) def test_startswith_concat(self): self.assert_compile( column('x').startswith('y'), "x LIKE concat(%s, '%%')", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_not_startswith_concat(self): self.assert_compile( ~column('x').startswith('y'), "x NOT LIKE concat(%s, '%%')", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_startswith_firebird(self): self.assert_compile( column('x').startswith('y'), "x STARTING WITH :x_1", checkparams={'x_1': 'y'}, dialect=firebird.dialect() ) def test_not_startswith_firebird(self): self.assert_compile( ~column('x').startswith('y'), "x NOT STARTING WITH :x_1", checkparams={'x_1': 'y'}, dialect=firebird.dialect() ) def test_startswith_literal_mysql(self): self.assert_compile( column('x').startswith(literal_column('y')), "x LIKE concat(y, '%%')", checkparams={}, dialect=mysql.dialect() ) def test_startswith_text_mysql(self): self.assert_compile( column('x').startswith(text('y')), "x LIKE concat(y, '%%')", checkparams={}, dialect=mysql.dialect() ) def test_endswith(self): self.assert_compile( column('x').endswith('y'), "x LIKE '%%' || :x_1", checkparams={'x_1': 'y'} ) def test_endswith_escape(self): self.assert_compile( column('x').endswith('y', escape='\\'), "x LIKE '%%' || :x_1 ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_not_endswith(self): self.assert_compile( ~column('x').endswith('y'), "x NOT LIKE '%%' || :x_1", checkparams={'x_1': 'y'} ) def test_not_endswith_escape(self): self.assert_compile( ~column('x').endswith('y', escape='\\'), "x NOT LIKE '%%' || :x_1 ESCAPE '\\'", checkparams={'x_1': 'y'} ) def test_endswith_literal(self): self.assert_compile( column('x').endswith(literal_column('y')), "x LIKE '%%' || y", checkparams={} ) def test_endswith_text(self): self.assert_compile( column('x').endswith(text('y')), "x LIKE '%%' || y", checkparams={} ) def test_endswith_mysql(self): self.assert_compile( column('x').endswith('y'), "x LIKE concat('%%', %s)", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_not_endswith_mysql(self): self.assert_compile( ~column('x').endswith('y'), "x NOT LIKE concat('%%', %s)", checkparams={'x_1': 'y'}, dialect=mysql.dialect() ) def test_endswith_literal_mysql(self): self.assert_compile( column('x').endswith(literal_column('y')), "x LIKE concat('%%', y)", checkparams={}, dialect=mysql.dialect() ) def test_endswith_text_mysql(self): self.assert_compile( column('x').endswith(text('y')), "x LIKE concat('%%', y)", checkparams={}, dialect=mysql.dialect() ) class CustomOpTest(fixtures.TestBase): def test_is_comparison(self): c = column('x') c2 = column('y') op1 = c.op('$', is_comparison=True)(c2).operator op2 = c.op('$', is_comparison=False)(c2).operator assert operators.is_comparison(op1) assert not operators.is_comparison(op2) class TupleTypingTest(fixtures.TestBase): def _assert_types(self, expr): eq_(expr.clauses[0].type._type_affinity, Integer) eq_(expr.clauses[1].type._type_affinity, String) eq_(expr.clauses[2].type._type_affinity, LargeBinary()._type_affinity) def test_type_coersion_on_eq(self): a, b, c = column( 'a', Integer), column( 'b', String), column( 'c', LargeBinary) t1 = tuple_(a, b, c) expr = t1 == (3, 'hi', 'there') self._assert_types(expr.right) def test_type_coersion_on_in(self): a, b, c = column( 'a', Integer), column( 'b', String), column( 'c', LargeBinary) t1 = tuple_(a, b, c) expr = t1.in_([(3, 'hi', 'there'), (4, 'Q', 'P')]) eq_(len(expr.right.clauses), 2) for elem in expr.right.clauses: self._assert_types(elem)