# coding: utf-8

import datetime

import decimal

import importlib

import operator

import os

import sqlalchemy as sa

from sqlalchemy import and_

from sqlalchemy import ARRAY

from sqlalchemy import BigInteger

from sqlalchemy import bindparam

from sqlalchemy import BLOB

from sqlalchemy import BOOLEAN

from sqlalchemy import Boolean

from sqlalchemy import cast

from sqlalchemy import CHAR

from sqlalchemy import CLOB

from sqlalchemy import DATE

from sqlalchemy import Date

from sqlalchemy import DATETIME

from sqlalchemy import DateTime

from sqlalchemy import DECIMAL

from sqlalchemy import dialects

from sqlalchemy import distinct

from sqlalchemy import Enum

from sqlalchemy import exc

from sqlalchemy import FLOAT

from sqlalchemy import Float

from sqlalchemy import func

from sqlalchemy import inspection

from sqlalchemy import INTEGER

from sqlalchemy import Integer

from sqlalchemy import Interval

from sqlalchemy import JSON

from sqlalchemy import LargeBinary

from sqlalchemy import literal

from sqlalchemy import MetaData

from sqlalchemy import NCHAR

from sqlalchemy import NUMERIC

from sqlalchemy import Numeric

from sqlalchemy import NVARCHAR

from sqlalchemy import PickleType

from sqlalchemy import REAL

from sqlalchemy import select

from sqlalchemy import SMALLINT

from sqlalchemy import SmallInteger

from sqlalchemy import String

from sqlalchemy import testing

from sqlalchemy import Text

from sqlalchemy import text

from sqlalchemy import TIME

from sqlalchemy import Time

from sqlalchemy import TIMESTAMP

from sqlalchemy import type_coerce

from sqlalchemy import TypeDecorator

from sqlalchemy import types

from sqlalchemy import Unicode

from sqlalchemy import util

from sqlalchemy import VARCHAR

from sqlalchemy.engine import default

from sqlalchemy.schema import AddConstraint

from sqlalchemy.schema import CheckConstraint

from sqlalchemy.sql import column

from sqlalchemy.sql import ddl

from sqlalchemy.sql import elements

from sqlalchemy.sql import null

from sqlalchemy.sql import operators

from sqlalchemy.sql import sqltypes

from sqlalchemy.sql import table

from sqlalchemy.sql import visitors

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import AssertsCompiledSQL

from sqlalchemy.testing import AssertsExecutionResults

from sqlalchemy.testing import engines

from sqlalchemy.testing import eq_

from sqlalchemy.testing import expect_warnings

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_

from sqlalchemy.testing import is_not_

from sqlalchemy.testing import mock

from sqlalchemy.testing import pickleable

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

from sqlalchemy.testing.util import picklers

from sqlalchemy.testing.util import round_decimal

from sqlalchemy.util import OrderedDict

from sqlalchemy.util import u

def _all_dialect_modules():

return [

importlib.import_module("sqlalchemy.dialects.%s" % d)

for d in dialects.__all__

if not d.startswith("_")

]

def _all_dialects():

return [d.base.dialect() for d in _all_dialect_modules()]

def _types_for_mod(mod):

for key in dir(mod):

typ = getattr(mod, key)

if not isinstance(typ, type) or not issubclass(typ, types.TypeEngine):

continue

yield typ

def _all_types(omit_special_types=False):

seen = set()

for typ in _types_for_mod(types):

if omit_special_types and typ in (

types.TypeDecorator,

types.TypeEngine,

types.Variant,

):

continue

if typ in seen:

continue

seen.add(typ)

yield typ

for dialect in _all_dialect_modules():

for typ in _types_for_mod(dialect):

if typ in seen:

continue

seen.add(typ)

yield typ

class AdaptTest(fixtures.TestBase):

@testing.combinations(((t,) for t in _types_for_mod(types)), id_="n")

def test_uppercase_importable(self, typ):

if typ.__name__ == typ.__name__.upper():

assert getattr(sa, typ.__name__) is typ

assert typ.__name__ in types.__all__

@testing.combinations(

((d.name, d) for d in _all_dialects()), argnames="dialect", id_="ia"

)

@testing.combinations(

(REAL(), "REAL"),

(FLOAT(), "FLOAT"),

(NUMERIC(), "NUMERIC"),

(DECIMAL(), "DECIMAL"),

(INTEGER(), "INTEGER"),

(SMALLINT(), "SMALLINT"),

(TIMESTAMP(), ("TIMESTAMP", "TIMESTAMP WITHOUT TIME ZONE")),

(DATETIME(), "DATETIME"),

(DATE(), "DATE"),

(TIME(), ("TIME", "TIME WITHOUT TIME ZONE")),

(CLOB(), "CLOB"),

(VARCHAR(10), ("VARCHAR(10)", "VARCHAR(10 CHAR)")),

(

NVARCHAR(10),

("NVARCHAR(10)", "NATIONAL VARCHAR(10)", "NVARCHAR2(10)"),

),

(CHAR(), "CHAR"),

(NCHAR(), ("NCHAR", "NATIONAL CHAR")),

(BLOB(), ("BLOB", "BLOB SUB_TYPE 0")),

(BOOLEAN(), ("BOOLEAN", "BOOL", "INTEGER")),

argnames="type_, expected",

id_="ra",

)

def test_uppercase_rendering(self, dialect, type_, expected):

"""Test that uppercase types from types.py always render as their

type.

As of SQLA 0.6, using an uppercase type means you want specifically

that type. If the database in use doesn't support that DDL, it (the DB

backend) should raise an error - it means you should be using a

lowercased (genericized) type.

"""

if isinstance(expected, str):

expected = (expected,)

try:

compiled = type_.compile(dialect=dialect)

except NotImplementedError:

return

assert compiled in expected, "%r matches none of %r for dialect %s" % (

compiled,

expected,

dialect.name,

)

assert (

str(types.to_instance(type_)) in expected

), "default str() of type %r not expected, %r" % (type_, expected)

def _adaptions():

for typ in _all_types(omit_special_types=True):

# up adapt from LowerCase to UPPERCASE,

# as well as to all non-sqltypes

up_adaptions = [typ] + typ.__subclasses__()

yield "%s.%s" % (

typ.__module__,

typ.__name__,

), False, typ, up_adaptions

for subcl in typ.__subclasses__():

if (

subcl is not typ

and typ is not TypeDecorator

and "sqlalchemy" in subcl.__module__

):

yield "%s.%s" % (

subcl.__module__,

subcl.__name__,

), True, subcl, [typ]

@testing.combinations(_adaptions(), id_="iaaa")

def test_adapt_method(self, is_down_adaption, typ, target_adaptions):

"""ensure all types have a working adapt() method,

which creates a distinct copy.

The distinct copy ensures that when we cache

the adapted() form of a type against the original

in a weak key dictionary, a cycle is not formed.

This test doesn't test type-specific arguments of

adapt() beyond their defaults.

"""

if issubclass(typ, ARRAY):

t1 = typ(String)

else:

t1 = typ()

for cls in target_adaptions:

if (is_down_adaption and issubclass(typ, sqltypes.Emulated)) or (

not is_down_adaption and issubclass(cls, sqltypes.Emulated)

):

continue

# print("ADAPT %s -> %s" % (t1.__class__, cls))

t2 = t1.adapt(cls)

assert t1 is not t2

if is_down_adaption:

t2, t1 = t1, t2

for k in t1.__dict__:

if k in (

"impl",

"_is_oracle_number",

"_create_events",

"create_constraint",

"inherit_schema",

"schema",

"metadata",

"name",

):

continue

# assert each value was copied, or that

# the adapted type has a more specific

# value than the original (i.e. SQL Server

# applies precision=24 for REAL)

assert (

getattr(t2, k) == t1.__dict__[k] or t1.__dict__[k] is None

)

eq_(t1.evaluates_none().should_evaluate_none, True)

def test_python_type(self):

eq_(types.Integer().python_type, int)

eq_(types.Numeric().python_type, decimal.Decimal)

eq_(types.Numeric(asdecimal=False).python_type, float)

eq_(types.LargeBinary().python_type, util.binary_type)

eq_(types.Float().python_type, float)

eq_(types.Interval().python_type, datetime.timedelta)

eq_(types.Date().python_type, datetime.date)

eq_(types.DateTime().python_type, datetime.datetime)

eq_(types.String().python_type, str)

eq_(types.Unicode().python_type, util.text_type)

eq_(types.Enum("one", "two", "three").python_type, str)

assert_raises(

NotImplementedError, lambda: types.TypeEngine().python_type

)

@testing.uses_deprecated()

@testing.combinations(*[(t,) for t in _all_types(omit_special_types=True)])

def test_repr(self, typ):

if issubclass(typ, ARRAY):

t1 = typ(String)

else:

t1 = typ()

repr(t1)

def test_adapt_constructor_copy_override_kw(self):

"""test that adapt() can accept kw args that override

the state of the original object.

This essentially is testing the behavior of util.constructor_copy().

"""

t1 = String(length=50)

t2 = t1.adapt(Text)

eq_(t2.length, 50)

def test_convert_unicode_text_type(self):

with testing.expect_deprecated(

"The String.convert_unicode parameter is deprecated"

):

eq_(types.String(convert_unicode=True).python_type, util.text_type)

class TypeAffinityTest(fixtures.TestBase):

@testing.combinations(

(String(), String),

(VARCHAR(), String),

(Date(), Date),

(LargeBinary(), types._Binary),

id_="rn",

)

def test_type_affinity(self, type_, affin):

eq_(type_._type_affinity, affin)

@testing.combinations(

(Integer(), SmallInteger(), True),

(Integer(), String(), False),

(Integer(), Integer(), True),

(Text(), String(), True),

(Text(), Unicode(), True),

(LargeBinary(), Integer(), False),

(LargeBinary(), PickleType(), True),

(PickleType(), LargeBinary(), True),

(PickleType(), PickleType(), True),

id_="rra",

)

def test_compare_type_affinity(self, t1, t2, comp):

eq_(t1._compare_type_affinity(t2), comp, "%s %s" % (t1, t2))

def test_decorator_doesnt_cache(self):

from sqlalchemy.dialects import postgresql

class MyType(TypeDecorator):

impl = CHAR

def load_dialect_impl(self, dialect):

if dialect.name == "postgresql":

return dialect.type_descriptor(postgresql.UUID())

else:

return dialect.type_descriptor(CHAR(32))

t1 = MyType()

d = postgresql.dialect()

assert t1._type_affinity is String

assert t1.dialect_impl(d)._type_affinity is postgresql.UUID

class PickleTypesTest(fixtures.TestBase):

@testing.combinations(

("Boo", Boolean()),

("Str", String()),

("Tex", Text()),

("Uni", Unicode()),

("Int", Integer()),

("Sma", SmallInteger()),

("Big", BigInteger()),

("Num", Numeric()),

("Flo", Float()),

("Dat", DateTime()),

("Dat", Date()),

("Tim", Time()),

("Lar", LargeBinary()),

("Pic", PickleType()),

("Int", Interval()),

id_="ar",

)

def test_pickle_types(self, name, type_):

column_type = Column(name, type_)

meta = MetaData()

Table("foo", meta, column_type)

for loads, dumps in picklers():

loads(dumps(column_type))

loads(dumps(meta))

class _UserDefinedTypeFixture(object):

@classmethod

def define_tables(cls, metadata):

class MyType(types.UserDefinedType):

def get_col_spec(self):

return "VARCHAR(100)"

def bind_processor(self, dialect):

def process(value):

return "BIND_IN" + value

return process

def result_processor(self, dialect, coltype):

def process(value):

return value + "BIND_OUT"

return process

def adapt(self, typeobj):

return typeobj()

class MyDecoratedType(types.TypeDecorator):

impl = String

def bind_processor(self, dialect):

impl_processor = super(MyDecoratedType, self).bind_processor(

dialect

) or (lambda value: value)

def process(value):

return "BIND_IN" + impl_processor(value)

return process

def result_processor(self, dialect, coltype):

impl_processor = super(MyDecoratedType, self).result_processor(

dialect, coltype

) or (lambda value: value)

def process(value):

return impl_processor(value) + "BIND_OUT"

return process

def copy(self):

return MyDecoratedType()

class MyNewUnicodeType(types.TypeDecorator):

impl = Unicode

def process_bind_param(self, value, dialect):

return "BIND_IN" + value

def process_result_value(self, value, dialect):

return value + "BIND_OUT"

def copy(self):

return MyNewUnicodeType(self.impl.length)

class MyNewIntType(types.TypeDecorator):

impl = Integer

def process_bind_param(self, value, dialect):

return value * 10

def process_result_value(self, value, dialect):

return value * 10

def copy(self):

return MyNewIntType()

class MyNewIntSubClass(MyNewIntType):

def process_result_value(self, value, dialect):

return value * 15

def copy(self):

return MyNewIntSubClass()

class MyUnicodeType(types.TypeDecorator):

impl = Unicode

def bind_processor(self, dialect):

impl_processor = super(MyUnicodeType, self).bind_processor(

dialect

) or (lambda value: value)

def process(value):

return "BIND_IN" + impl_processor(value)

return process

def result_processor(self, dialect, coltype):

impl_processor = super(MyUnicodeType, self).result_processor(

dialect, coltype

) or (lambda value: value)

def process(value):

return impl_processor(value) + "BIND_OUT"

return process

def copy(self):

return MyUnicodeType(self.impl.length)

Table(

"users",

metadata,

Column("user_id", Integer, primary_key=True),

# totall custom type

Column("goofy", MyType, nullable=False),

# decorated type with an argument, so its a String

Column("goofy2", MyDecoratedType(50), nullable=False),

Column("goofy4", MyUnicodeType(50), nullable=False),

Column("goofy7", MyNewUnicodeType(50), nullable=False),

Column("goofy8", MyNewIntType, nullable=False),

Column("goofy9", MyNewIntSubClass, nullable=False),

)

class UserDefinedRoundTripTest(_UserDefinedTypeFixture, fixtures.TablesTest):

__backend__ = True

def _data_fixture(self):

users = self.tables.users

with testing.db.connect() as conn:

conn.execute(

users.insert(),

dict(

user_id=2,

goofy="jack",

goofy2="jack",

goofy4=util.u("jack"),

goofy7=util.u("jack"),

goofy8=12,

goofy9=12,

),

)

conn.execute(

users.insert(),

dict(

user_id=3,

goofy="lala",

goofy2="lala",

goofy4=util.u("lala"),

goofy7=util.u("lala"),

goofy8=15,

goofy9=15,

),

)

conn.execute(

users.insert(),

dict(

user_id=4,

goofy="fred",

goofy2="fred",

goofy4=util.u("fred"),

goofy7=util.u("fred"),

goofy8=9,

goofy9=9,

),

)

def test_processing(self, connection):

users = self.tables.users

self._data_fixture()

result = connection.execute(

users.select().order_by(users.c.user_id)

).fetchall()

for assertstr, assertint, assertint2, row in zip(

[

"BIND_INjackBIND_OUT",

"BIND_INlalaBIND_OUT",

"BIND_INfredBIND_OUT",

],

[1200, 1500, 900],

[1800, 2250, 1350],

result,

):

for col in list(row)[1:5]:

eq_(col, assertstr)

eq_(row[5], assertint)

eq_(row[6], assertint2)

for col in row[3], row[4]:

assert isinstance(col, util.text_type)

def test_plain_in(self, connection):

users = self.tables.users

self._data_fixture()

stmt = (

select([users.c.user_id, users.c.goofy8])

.where(users.c.goofy8.in_([15, 9]))

.order_by(users.c.user_id)

)

result = connection.execute(stmt, {"goofy": [15, 9]})

eq_(result.fetchall(), [(3, 1500), (4, 900)])

def test_expanding_in(self, connection):

users = self.tables.users

self._data_fixture()

stmt = (

select([users.c.user_id, users.c.goofy8])

.where(users.c.goofy8.in_(bindparam("goofy", expanding=True)))

.order_by(users.c.user_id)

)

result = connection.execute(stmt, {"goofy": [15, 9]})

eq_(result.fetchall(), [(3, 1500), (4, 900)])

class UserDefinedTest(

_UserDefinedTypeFixture, fixtures.TablesTest, AssertsCompiledSQL

):

run_create_tables = None

run_inserts = None

run_deletes = None

"""tests user-defined types."""

def test_typedecorator_literal_render(self):

class MyType(types.TypeDecorator):

impl = String

def process_literal_param(self, value, dialect):

return "HI->%s<-THERE" % value

self.assert_compile(

select([literal("test", MyType)]),

"SELECT 'HI->test<-THERE' AS anon_1",

dialect="default",

literal_binds=True,

)

def test_kw_colspec(self):

class MyType(types.UserDefinedType):

def get_col_spec(self, **kw):

return "FOOB %s" % kw["type_expression"].name

class MyOtherType(types.UserDefinedType):

def get_col_spec(self):

return "BAR"

t = Table("t", MetaData(), Column("bar", MyType, nullable=False))

self.assert_compile(ddl.CreateColumn(t.c.bar), "bar FOOB bar NOT NULL")

t = Table("t", MetaData(), Column("bar", MyOtherType, nullable=False))

self.assert_compile(ddl.CreateColumn(t.c.bar), "bar BAR NOT NULL")

def test_typedecorator_literal_render_fallback_bound(self):

# fall back to process_bind_param for literal

# value rendering.

class MyType(types.TypeDecorator):

impl = String

def process_bind_param(self, value, dialect):

return "HI->%s<-THERE" % value

self.assert_compile(

select([literal("test", MyType)]),

"SELECT 'HI->test<-THERE' AS anon_1",

dialect="default",

literal_binds=True,

)

def test_typedecorator_impl(self):

for impl_, exp, kw in [

(Float, "FLOAT", {}),

(Float, "FLOAT(2)", {"precision": 2}),

(Float(2), "FLOAT(2)", {"precision": 4}),

(Numeric(19, 2), "NUMERIC(19, 2)", {}),

]:

for dialect_ in (

dialects.postgresql,

dialects.mssql,

dialects.mysql,

):

dialect_ = dialect_.dialect()

raw_impl = types.to_instance(impl_, **kw)

class MyType(types.TypeDecorator):

impl = impl_

dec_type = MyType(**kw)

eq_(dec_type.impl.__class__, raw_impl.__class__)

raw_dialect_impl = raw_impl.dialect_impl(dialect_)

dec_dialect_impl = dec_type.dialect_impl(dialect_)

eq_(dec_dialect_impl.__class__, MyType)

eq_(

raw_dialect_impl.__class__, dec_dialect_impl.impl.__class__

)

self.assert_compile(MyType(**kw), exp, dialect=dialect_)

def test_user_defined_typedec_impl(self):

class MyType(types.TypeDecorator):

impl = Float

def load_dialect_impl(self, dialect):

if dialect.name == "sqlite":

return String(50)

else:

return super(MyType, self).load_dialect_impl(dialect)

sl = dialects.sqlite.dialect()

pg = dialects.postgresql.dialect()

t = MyType()

self.assert_compile(t, "VARCHAR(50)", dialect=sl)

self.assert_compile(t, "FLOAT", dialect=pg)

eq_(

t.dialect_impl(dialect=sl).impl.__class__,

String().dialect_impl(dialect=sl).__class__,

)

eq_(

t.dialect_impl(dialect=pg).impl.__class__,

Float().dialect_impl(pg).__class__,

)

def test_type_decorator_repr(self):

class MyType(TypeDecorator):

impl = VARCHAR

eq_(repr(MyType(45)), "MyType(length=45)")

def test_user_defined_typedec_impl_bind(self):

class TypeOne(types.TypeEngine):

def bind_processor(self, dialect):

def go(value):

return value + " ONE"

return go

class TypeTwo(types.TypeEngine):

def bind_processor(self, dialect):

def go(value):

return value + " TWO"

return go

class MyType(types.TypeDecorator):

impl = TypeOne

def load_dialect_impl(self, dialect):

if dialect.name == "sqlite":

return TypeOne()

else:

return TypeTwo()

def process_bind_param(self, value, dialect):

return "MYTYPE " + value

sl = dialects.sqlite.dialect()

pg = dialects.postgresql.dialect()

t = MyType()

eq_(t._cached_bind_processor(sl)("foo"), "MYTYPE foo ONE")

eq_(t._cached_bind_processor(pg)("foo"), "MYTYPE foo TWO")

def test_user_defined_dialect_specific_args(self):

class MyType(types.UserDefinedType):

def __init__(self, foo="foo", **kwargs):

super(MyType, self).__init__()

self.foo = foo

self.dialect_specific_args = kwargs

def adapt(self, cls):

return cls(foo=self.foo, **self.dialect_specific_args)

t = MyType(bar="bar")

a = t.dialect_impl(testing.db.dialect)

eq_(a.foo, "foo")

eq_(a.dialect_specific_args["bar"], "bar")

class StringConvertUnicodeTest(fixtures.TestBase):

@testing.combinations((Unicode,), (String,), argnames="datatype")

@testing.combinations((True,), (False,), argnames="convert_unicode")

@testing.combinations(

(String.RETURNS_CONDITIONAL,),

(String.RETURNS_BYTES,),

(String.RETURNS_UNICODE),

argnames="returns_unicode_strings",

)

def test_convert_unicode(

self, datatype, convert_unicode, returns_unicode_strings

):

s1 = datatype()

dialect = mock.Mock(

returns_unicode_strings=returns_unicode_strings,

encoding="utf-8",

convert_unicode=convert_unicode,

)

proc = s1.result_processor(dialect, None)

string = u("méil")

bytestring = string.encode("utf-8")

if (

datatype is Unicode or convert_unicode

) and returns_unicode_strings in (

String.RETURNS_CONDITIONAL,

String.RETURNS_BYTES,

):

eq_(proc(bytestring), string)

if returns_unicode_strings is String.RETURNS_CONDITIONAL:

eq_(proc(string), string)

else:

if util.py3k:

# trying to decode a unicode

assert_raises(TypeError, proc, string)

else:

assert_raises(UnicodeEncodeError, proc, string)

else:

is_(proc, None)

class TypeCoerceCastTest(fixtures.TablesTest):

__backend__ = True

@classmethod

def define_tables(cls, metadata):

class MyType(types.TypeDecorator):

impl = String(50)

def process_bind_param(self, value, dialect):

return "BIND_IN" + str(value)

def process_result_value(self, value, dialect):

return value + "BIND_OUT"

cls.MyType = MyType

Table("t", metadata, Column("data", String(50)))

def test_insert_round_trip_cast(self, connection):

self._test_insert_round_trip(cast, connection)

def test_insert_round_trip_type_coerce(self, connection):

self._test_insert_round_trip(type_coerce, connection)

def _test_insert_round_trip(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

eq_(

conn.execute(select([coerce_fn(t.c.data, MyType)])).fetchall(),

[("BIND_INd1BIND_OUT",)],

)

def test_coerce_from_nulltype_cast(self, connection):

self._test_coerce_from_nulltype(cast, connection)

def test_coerce_from_nulltype_type_coerce(self, connection):

self._test_coerce_from_nulltype(type_coerce, connection)

def _test_coerce_from_nulltype(self, coerce_fn, conn):

MyType = self.MyType

# test coerce from nulltype - e.g. use an object that

# does't match to a known type

class MyObj(object):

def __str__(self):

return "THISISMYOBJ"

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn(MyObj(), MyType)))

eq_(

conn.execute(select([coerce_fn(t.c.data, MyType)])).fetchall(),

[("BIND_INTHISISMYOBJBIND_OUT",)],

)

def test_vs_non_coerced_cast(self, connection):

self._test_vs_non_coerced(cast, connection)

def test_vs_non_coerced_type_coerce(self, connection):

self._test_vs_non_coerced(type_coerce, connection)

def _test_vs_non_coerced(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

eq_(

conn.execute(

select([t.c.data, coerce_fn(t.c.data, MyType)])

).fetchall(),

[("BIND_INd1", "BIND_INd1BIND_OUT")],

)

def test_vs_non_coerced_alias_cast(self, connection):

self._test_vs_non_coerced_alias(cast, connection)

def test_vs_non_coerced_alias_type_coerce(self, connection):

self._test_vs_non_coerced_alias(type_coerce, connection)

def _test_vs_non_coerced_alias(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

eq_(

conn.execute(

select([t.c.data.label("x"), coerce_fn(t.c.data, MyType)])

.alias()

.select()

).fetchall(),

[("BIND_INd1", "BIND_INd1BIND_OUT")],

)

def test_vs_non_coerced_where_cast(self, connection):

self._test_vs_non_coerced_where(cast, connection)

def test_vs_non_coerced_where_type_coerce(self, connection):

self._test_vs_non_coerced_where(type_coerce, connection)

def _test_vs_non_coerced_where(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

# coerce on left side

eq_(

conn.execute(

select([t.c.data, coerce_fn(t.c.data, MyType)]).where(

coerce_fn(t.c.data, MyType) == "d1"

)

).fetchall(),

[("BIND_INd1", "BIND_INd1BIND_OUT")],

)

# coerce on right side

eq_(

conn.execute(

select([t.c.data, coerce_fn(t.c.data, MyType)]).where(

t.c.data == coerce_fn("d1", MyType)

)

).fetchall(),

[("BIND_INd1", "BIND_INd1BIND_OUT")],

)

def test_coerce_none_cast(self, connection):

self._test_coerce_none(cast, connection)

def test_coerce_none_type_coerce(self, connection):

self._test_coerce_none(type_coerce, connection)

def _test_coerce_none(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

eq_(

conn.execute(

select([t.c.data, coerce_fn(t.c.data, MyType)]).where(

t.c.data == coerce_fn(None, MyType)

)

).fetchall(),

[],

)

eq_(

conn.execute(

select([t.c.data, coerce_fn(t.c.data, MyType)]).where(

coerce_fn(t.c.data, MyType) == None

)

).fetchall(), # noqa

[],

)

def test_resolve_clause_element_cast(self, connection):

self._test_resolve_clause_element(cast, connection)

def test_resolve_clause_element_type_coerce(self, connection):

self._test_resolve_clause_element(type_coerce, connection)

def _test_resolve_clause_element(self, coerce_fn, conn):

MyType = self.MyType

t = self.tables.t

conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))

class MyFoob(object):

def __clause_element__(self):

return t.c.data

eq_(

conn.execute(

select([t.c.data, coerce_fn(MyFoob(), MyType)])

).fetchall(),

[("BIND_INd1", "BIND_INd1BIND_OUT")],

)

def test_cast_replace_col_w_bind(self, connection):

self._test_replace_col_w_bind(cast, connection)

def test_type_coerce_replace_col_w_bind(self, connection):

self._test_replace_col_w_bind(type_coerce, connection)

def _test_replace_col_w_bind(self, coerce_fn, conn):