# coding: utf-8

from contextlib import contextmanager

import re

import weakref

import sqlalchemy as tsa

from sqlalchemy import bindparam

from sqlalchemy import create_engine

from sqlalchemy import create_mock_engine

from sqlalchemy import event

from sqlalchemy import func

from sqlalchemy import inspect

from sqlalchemy import INT

from sqlalchemy import Integer

from sqlalchemy import LargeBinary

from sqlalchemy import MetaData

from sqlalchemy import select

from sqlalchemy import Sequence

from sqlalchemy import String

from sqlalchemy import testing

from sqlalchemy import text

from sqlalchemy import TypeDecorator

from sqlalchemy import util

from sqlalchemy import VARCHAR

from sqlalchemy.engine import default

from sqlalchemy.engine.base import Connection

from sqlalchemy.engine.base import Engine

from sqlalchemy.sql import column

from sqlalchemy.sql import literal

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import config

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_false

from sqlalchemy.testing import is_not_

from sqlalchemy.testing import is_true

from sqlalchemy.testing import mock

from sqlalchemy.testing.assertsql import CompiledSQL

from sqlalchemy.testing.engines import testing_engine

from sqlalchemy.testing.mock import call

from sqlalchemy.testing.mock import Mock

from sqlalchemy.testing.mock import patch

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

from sqlalchemy.testing.util import gc_collect

from sqlalchemy.testing.util import picklers

from sqlalchemy.util import collections_abc

class SomeException(Exception):

pass

class ExecuteTest(fixtures.TablesTest):

__backend__ = True

@classmethod

def define_tables(cls, metadata):

Table(

"users",

metadata,

Column("user_id", INT, primary_key=True, autoincrement=False),

Column("user_name", VARCHAR(20)),

)

Table(

"users_autoinc",

metadata,

Column(

"user_id", INT, primary_key=True, test_needs_autoincrement=True

),

Column("user_name", VARCHAR(20)),

)

@testing.fails_on(

"postgresql+pg8000",

"pg8000 still doesn't allow single paren without params",

)

def test_no_params_option(self):

stmt = (

"SELECT '%'"

+ testing.db.dialect.statement_compiler(

testing.db.dialect, None

).default_from()

)

conn = testing.db.connect()

result = (

conn.execution_options(no_parameters=True)

.exec_driver_sql(stmt)

.scalar()

)

eq_(result, "%")

def test_raw_positional_invalid(self, connection):

assert_raises_message(

tsa.exc.ArgumentError,

"List argument must consist only of tuples or dictionaries",

connection.exec_driver_sql,

"insert into users (user_id, user_name) " "values (?, ?)",

[2, "fred"],

)

assert_raises_message(

tsa.exc.ArgumentError,

"List argument must consist only of tuples or dictionaries",

connection.exec_driver_sql,

"insert into users (user_id, user_name) " "values (?, ?)",

[[3, "ed"], [4, "horse"]],

)

def test_raw_named_invalid(self, connection):

# this is awkward b.c. this is just testing if regular Python

# is raising TypeError if they happened to send arguments that

# look like the legacy ones which also happen to conflict with

# the positional signature for the method. some combinations

# can get through and fail differently

assert_raises(

TypeError,

connection.exec_driver_sql,

"insert into users (user_id, user_name) "

"values (%(id)s, %(name)s)",

{"id": 2, "name": "ed"},

{"id": 3, "name": "horse"},

{"id": 4, "name": "horse"},

)

assert_raises(

TypeError,

connection.exec_driver_sql,

"insert into users (user_id, user_name) "

"values (%(id)s, %(name)s)",

id=4,

name="sally",

)

@testing.requires.qmark_paramstyle

def test_raw_qmark(self, connection):

conn = connection

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (?, ?)",

(1, "jack"),

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (?, ?)",

(2, "fred"),

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (?, ?)",

[(3, "ed"), (4, "horse")],

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (?, ?)",

[(5, "barney"), (6, "donkey")],

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (?, ?)",

(7, "sally"),

)

res = conn.exec_driver_sql("select * from users order by user_id")

assert res.fetchall() == [

(1, "jack"),

(2, "fred"),

(3, "ed"),

(4, "horse"),

(5, "barney"),

(6, "donkey"),

(7, "sally"),

]

res = conn.exec_driver_sql(

"select * from users where user_name=?", ("jack",)

)

assert res.fetchall() == [(1, "jack")]

@testing.requires.format_paramstyle

def test_raw_sprintf(self, connection):

conn = connection

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (%s, %s)",

(1, "jack"),

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (%s, %s)",

[(2, "ed"), (3, "horse")],

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (%s, %s)",

(4, "sally"),

)

conn.exec_driver_sql("insert into users (user_id) values (%s)", (5,))

res = conn.exec_driver_sql("select * from users order by user_id")

assert res.fetchall() == [

(1, "jack"),

(2, "ed"),

(3, "horse"),

(4, "sally"),

(5, None),

]

res = conn.exec_driver_sql(

"select * from users where user_name=%s", ("jack",)

)

assert res.fetchall() == [(1, "jack")]

@testing.requires.pyformat_paramstyle

def test_raw_python(self, connection):

conn = connection

conn.exec_driver_sql(

"insert into users (user_id, user_name) "

"values (%(id)s, %(name)s)",

{"id": 1, "name": "jack"},

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) "

"values (%(id)s, %(name)s)",

[{"id": 2, "name": "ed"}, {"id": 3, "name": "horse"}],

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) "

"values (%(id)s, %(name)s)",

dict(id=4, name="sally"),

)

res = conn.exec_driver_sql("select * from users order by user_id")

assert res.fetchall() == [

(1, "jack"),

(2, "ed"),

(3, "horse"),

(4, "sally"),

]

@testing.requires.named_paramstyle

def test_raw_named(self, connection):

conn = connection

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (:id, :name)",

{"id": 1, "name": "jack"},

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (:id, :name)",

[{"id": 2, "name": "ed"}, {"id": 3, "name": "horse"}],

)

conn.exec_driver_sql(

"insert into users (user_id, user_name) " "values (:id, :name)",

{"id": 4, "name": "sally"},

)

res = conn.exec_driver_sql("select * from users order by user_id")

assert res.fetchall() == [

(1, "jack"),

(2, "ed"),

(3, "horse"),

(4, "sally"),

]

@testing.engines.close_open_connections

def test_exception_wrapping_dbapi(self):

conn = testing.db.connect()

# engine does not have exec_driver_sql

assert_raises_message(

tsa.exc.DBAPIError,

r"not_a_valid_statement",

conn.exec_driver_sql,

"not_a_valid_statement",

)

@testing.requires.sqlite

def test_exception_wrapping_non_dbapi_error(self):

e = create_engine("sqlite://")

e.dialect.is_disconnect = is_disconnect = Mock()

with e.connect() as c:

c.connection.cursor = Mock(

return_value=Mock(

execute=Mock(

side_effect=TypeError("I'm not a DBAPI error")

)

)

)

assert_raises_message(

TypeError,

"I'm not a DBAPI error",

c.exec_driver_sql,

"select ",

)

eq_(is_disconnect.call_count, 0)

def test_exception_wrapping_non_standard_dbapi_error(self):

class DBAPIError(Exception):

pass

class OperationalError(DBAPIError):

pass

class NonStandardException(OperationalError):

pass

with patch.object(

testing.db.dialect, "dbapi", Mock(Error=DBAPIError)

), patch.object(

testing.db.dialect, "is_disconnect", lambda *arg: False

), patch.object(

testing.db.dialect,

"do_execute",

Mock(side_effect=NonStandardException),

):

with testing.db.connect() as conn:

assert_raises(

tsa.exc.OperationalError, conn.exec_driver_sql, "select 1"

)

def test_exception_wrapping_non_dbapi_statement(self):

class MyType(TypeDecorator):

impl = Integer

def process_bind_param(self, value, dialect):

raise SomeException("nope")

def _go(conn):

assert_raises_message(

tsa.exc.StatementError,

r"\(.*.SomeException\) " r"nope\n\[SQL\: u?SELECT 1 ",

conn.execute,

select([1]).where(column("foo") == literal("bar", MyType())),

)

_go(testing.db)

with testing.db.connect() as conn:

_go(conn)

def test_not_an_executable(self):

for obj in (

Table("foo", MetaData(), Column("x", Integer)),

Column("x", Integer),

tsa.and_(True),

tsa.and_(True).compile(),

column("foo"),

column("foo").compile(),

MetaData(),

Integer(),

tsa.Index(name="foo"),

tsa.UniqueConstraint("x"),

):

with testing.db.connect() as conn:

assert_raises_message(

tsa.exc.ObjectNotExecutableError,

"Not an executable object",

conn.execute,

obj,

)

def test_stmt_exception_bytestring_raised(self):

name = util.u("méil")

with testing.db.connect() as conn:

assert_raises_message(

tsa.exc.StatementError,

util.u(

"A value is required for bind parameter 'uname'\n"

r".*SELECT users.user_name AS .m\xe9il."

)

if util.py2k

else util.u(

"A value is required for bind parameter 'uname'\n"

".*SELECT users.user_name AS .méil."

),

conn.execute,

select([users.c.user_name.label(name)]).where(

users.c.user_name == bindparam("uname")

),

{"uname_incorrect": "foo"},

)

def test_stmt_exception_bytestring_utf8(self):

# uncommon case for Py3K, bytestring object passed

# as the error message

message = util.u("some message méil").encode("utf-8")

err = tsa.exc.SQLAlchemyError(message)

if util.py2k:

# string passes it through

eq_(str(err), message)

# unicode accessor decodes to utf-8

eq_(unicode(err), util.u("some message méil")) # noqa

else:

eq_(str(err), util.u("some message méil"))

def test_stmt_exception_bytestring_latin1(self):

# uncommon case for Py3K, bytestring object passed

# as the error message

message = util.u("some message méil").encode("latin-1")

err = tsa.exc.SQLAlchemyError(message)

if util.py2k:

# string passes it through

eq_(str(err), message)

# unicode accessor decodes to utf-8

eq_(unicode(err), util.u("some message m\\xe9il")) # noqa

else:

eq_(str(err), util.u("some message m\\xe9il"))

def test_stmt_exception_unicode_hook_unicode(self):

# uncommon case for Py2K, Unicode object passed

# as the error message

message = util.u("some message méil")

err = tsa.exc.SQLAlchemyError(message)

if util.py2k:

eq_(unicode(err), util.u("some message méil")) # noqa

else:

eq_(str(err), util.u("some message méil"))

def test_stmt_exception_str_multi_args(self):

err = tsa.exc.SQLAlchemyError("some message", 206)

eq_(str(err), "('some message', 206)")

def test_stmt_exception_str_multi_args_bytestring(self):

message = util.u("some message méil").encode("utf-8")

err = tsa.exc.SQLAlchemyError(message, 206)

eq_(str(err), str((message, 206)))

def test_stmt_exception_str_multi_args_unicode(self):

message = util.u("some message méil")

err = tsa.exc.SQLAlchemyError(message, 206)

eq_(str(err), str((message, 206)))

def test_stmt_exception_pickleable_no_dbapi(self):

self._test_stmt_exception_pickleable(Exception("hello world"))

@testing.crashes(

"postgresql+psycopg2",

"Older versions don't support cursor pickling, newer ones do",

)

@testing.fails_on(

"mysql+oursql",

"Exception doesn't come back exactly the same from pickle",

)

@testing.fails_on(

"mysql+mysqlconnector",

"Exception doesn't come back exactly the same from pickle",

)

@testing.fails_on(

"oracle+cx_oracle",

"cx_oracle exception seems to be having " "some issue with pickling",

)

def test_stmt_exception_pickleable_plus_dbapi(self):

raw = testing.db.raw_connection()

the_orig = None

try:

try:

cursor = raw.cursor()

cursor.execute("SELECTINCORRECT")

except testing.db.dialect.dbapi.DatabaseError as orig:

# py3k has "orig" in local scope...

the_orig = orig

finally:

raw.close()

self._test_stmt_exception_pickleable(the_orig)

def _test_stmt_exception_pickleable(self, orig):

for sa_exc in (

tsa.exc.StatementError(

"some error",

"select * from table",

{"foo": "bar"},

orig,

False,

),

tsa.exc.InterfaceError(

"select * from table", {"foo": "bar"}, orig, True

),

tsa.exc.NoReferencedTableError("message", "tname"),

tsa.exc.NoReferencedColumnError("message", "tname", "cname"),

tsa.exc.CircularDependencyError(

"some message", [1, 2, 3], [(1, 2), (3, 4)]

),

):

for loads, dumps in picklers():

repickled = loads(dumps(sa_exc))

eq_(repickled.args[0], sa_exc.args[0])

if isinstance(sa_exc, tsa.exc.StatementError):

eq_(repickled.params, {"foo": "bar"})

eq_(repickled.statement, sa_exc.statement)

if hasattr(sa_exc, "connection_invalidated"):

eq_(

repickled.connection_invalidated,

sa_exc.connection_invalidated,

)

eq_(repickled.orig.args[0], orig.args[0])

def test_dont_wrap_mixin(self):

class MyException(Exception, tsa.exc.DontWrapMixin):

pass

class MyType(TypeDecorator):

impl = Integer

def process_bind_param(self, value, dialect):

raise MyException("nope")

def _go(conn):

assert_raises_message(

MyException,

"nope",

conn.execute,

select([1]).where(column("foo") == literal("bar", MyType())),

)

_go(testing.db)

conn = testing.db.connect()

try:

_go(conn)

finally:

conn.close()

def test_empty_insert(self):

"""test that execute() interprets [] as a list with no params"""

users_autoinc = self.tables.users_autoinc

testing.db.execute(

users_autoinc.insert().values(user_name=bindparam("name", None)),

[],

)

eq_(testing.db.execute(users_autoinc.select()).fetchall(), [(1, None)])

@testing.only_on("sqlite")

def test_execute_compiled_favors_compiled_paramstyle(self):

with patch.object(testing.db.dialect, "do_execute") as do_exec:

stmt = users.update().values(user_id=1, user_name="foo")

d1 = default.DefaultDialect(paramstyle="format")

d2 = default.DefaultDialect(paramstyle="pyformat")

testing.db.execute(stmt.compile(dialect=d1))

testing.db.execute(stmt.compile(dialect=d2))

eq_(

do_exec.mock_calls,

[

call(

mock.ANY,

"UPDATE users SET user_id=%s, user_name=%s",

(1, "foo"),

mock.ANY,

),

call(

mock.ANY,

"UPDATE users SET user_id=%(user_id)s, "

"user_name=%(user_name)s",

{"user_name": "foo", "user_id": 1},

mock.ANY,

),

],

)

@testing.requires.ad_hoc_engines

def test_engine_level_options(self):

eng = engines.testing_engine(

options={"execution_options": {"foo": "bar"}}

)

with eng.connect() as conn:

eq_(conn._execution_options["foo"], "bar")

eq_(

conn.execution_options(bat="hoho")._execution_options["foo"],

"bar",

)

eq_(

conn.execution_options(bat="hoho")._execution_options["bat"],

"hoho",

)

eq_(

conn.execution_options(foo="hoho")._execution_options["foo"],

"hoho",

)

eng.update_execution_options(foo="hoho")

conn = eng.connect()

eq_(conn._execution_options["foo"], "hoho")

@testing.requires.ad_hoc_engines

def test_generative_engine_execution_options(self):

eng = engines.testing_engine(

options={"execution_options": {"base": "x1"}}

)

is_(eng.engine, eng)

eng1 = eng.execution_options(foo="b1")

is_(eng1.engine, eng1)

eng2 = eng.execution_options(foo="b2")

eng1a = eng1.execution_options(bar="a1")

eng2a = eng2.execution_options(foo="b3", bar="a2")

is_(eng2a.engine, eng2a)

eq_(eng._execution_options, {"base": "x1"})

eq_(eng1._execution_options, {"base": "x1", "foo": "b1"})

eq_(eng2._execution_options, {"base": "x1", "foo": "b2"})

eq_(eng1a._execution_options, {"base": "x1", "foo": "b1", "bar": "a1"})

eq_(eng2a._execution_options, {"base": "x1", "foo": "b3", "bar": "a2"})

is_(eng1a.pool, eng.pool)

# test pool is shared

eng2.dispose()

is_(eng1a.pool, eng2.pool)

is_(eng.pool, eng2.pool)

@testing.requires.ad_hoc_engines

def test_autocommit_option_no_issue_first_connect(self):

eng = create_engine(testing.db.url)

eng.update_execution_options(autocommit=True)

conn = eng.connect()

eq_(conn._execution_options, {"autocommit": True})

conn.close()

def test_initialize_rollback(self):

"""test a rollback happens during first connect"""

eng = create_engine(testing.db.url)

with patch.object(eng.dialect, "do_rollback") as do_rollback:

assert do_rollback.call_count == 0

connection = eng.connect()

assert do_rollback.call_count == 1

connection.close()

@testing.requires.ad_hoc_engines

def test_dialect_init_uses_options(self):

eng = create_engine(testing.db.url)

def my_init(connection):

connection.execution_options(foo="bar").execute(select([1]))

with patch.object(eng.dialect, "initialize", my_init):

conn = eng.connect()

eq_(conn._execution_options, {})

conn.close()

@testing.requires.ad_hoc_engines

def test_generative_engine_event_dispatch_hasevents(self):

def l1(*arg, **kw):

pass

eng = create_engine(testing.db.url)

assert not eng._has_events

event.listen(eng, "before_execute", l1)

eng2 = eng.execution_options(foo="bar")

assert eng2._has_events

def test_works_after_dispose(self):

eng = create_engine(testing.db.url)

for i in range(3):

eq_(eng.scalar(select([1])), 1)

eng.dispose()

def test_works_after_dispose_testing_engine(self):

eng = engines.testing_engine()

for i in range(3):

eq_(eng.scalar(select([1])), 1)

eng.dispose()

class UnicodeReturnsTest(fixtures.TestBase):

@testing.requires.python3

def test_unicode_test_not_in_python3(self):

eng = engines.testing_engine()

eng.dialect.returns_unicode_strings = String.RETURNS_UNKNOWN

assert_raises_message(

tsa.exc.InvalidRequestError,

"RETURNS_UNKNOWN is unsupported in Python 3",

eng.connect,

)

@testing.requires.python2

def test_unicode_test_fails_warning(self):

class MockCursor(engines.DBAPIProxyCursor):

def execute(self, stmt, params=None, **kw):

if "test unicode returns" in stmt:

raise self.engine.dialect.dbapi.DatabaseError("boom")

else:

return super(MockCursor, self).execute(stmt, params, **kw)

eng = engines.proxying_engine(cursor_cls=MockCursor)

with testing.expect_warnings(

"Exception attempting to detect unicode returns"

):

eng.connect()

# because plain varchar passed, we don't know the correct answer

eq_(eng.dialect.returns_unicode_strings, String.RETURNS_CONDITIONAL)

eng.dispose()

class ConvenienceExecuteTest(fixtures.TablesTest):

__backend__ = True

@classmethod

def define_tables(cls, metadata):

cls.table = Table(

"exec_test",

metadata,

Column("a", Integer),

Column("b", Integer),

test_needs_acid=True,

)

def _trans_fn(self, is_transaction=False):

def go(conn, x, value=None):

if is_transaction:

conn = conn.connection

conn.execute(self.table.insert().values(a=x, b=value))

return go

def _trans_rollback_fn(self, is_transaction=False):

def go(conn, x, value=None):

if is_transaction:

conn = conn.connection

conn.execute(self.table.insert().values(a=x, b=value))

raise SomeException("breakage")

return go

def _assert_no_data(self):

eq_(

testing.db.scalar(

select([func.count("*")]).select_from(self.table)

),

0,

)

def _assert_fn(self, x, value=None):

eq_(testing.db.execute(self.table.select()).fetchall(), [(x, value)])

def test_transaction_engine_ctx_commit(self):

fn = self._trans_fn()

ctx = testing.db.begin()

testing.run_as_contextmanager(ctx, fn, 5, value=8)

self._assert_fn(5, value=8)

def test_transaction_engine_ctx_begin_fails(self):

engine = engines.testing_engine()

mock_connection = Mock(

return_value=Mock(begin=Mock(side_effect=Exception("boom")))

)

engine._connection_cls = mock_connection

assert_raises(Exception, engine.begin)

eq_(mock_connection.return_value.close.mock_calls, [call()])

def test_transaction_engine_ctx_rollback(self):

fn = self._trans_rollback_fn()

ctx = testing.db.begin()

assert_raises_message(

Exception,

"breakage",

testing.run_as_contextmanager,

ctx,

fn,

5,

value=8,

)

self._assert_no_data()

def test_transaction_connection_ctx_commit(self):

fn = self._trans_fn(True)

with testing.db.connect() as conn:

ctx = conn.begin()

testing.run_as_contextmanager(ctx, fn, 5, value=8)

self._assert_fn(5, value=8)

def test_transaction_connection_ctx_rollback(self):

fn = self._trans_rollback_fn(True)

with testing.db.connect() as conn:

ctx = conn.begin()

assert_raises_message(

Exception,

"breakage",

testing.run_as_contextmanager,

ctx,

fn,

5,

value=8,

)

self._assert_no_data()

def test_connection_as_ctx(self):

fn = self._trans_fn()

ctx = testing.db.connect()

testing.run_as_contextmanager(ctx, fn, 5, value=8)

# autocommit is on

self._assert_fn(5, value=8)

@testing.fails_on("mysql+oursql", "oursql bug ? getting wrong rowcount")

def test_connect_as_ctx_noautocommit(self):

fn = self._trans_fn()

self._assert_no_data()

with testing.db.connect() as conn:

ctx = conn.execution_options(autocommit=False)

testing.run_as_contextmanager(ctx, fn, 5, value=8)

# autocommit is off

self._assert_no_data()

class CompiledCacheTest(fixtures.TestBase):

__backend__ = True

@classmethod

def setup_class(cls):

global users, metadata

metadata = MetaData(testing.db)

users = Table(

"users",

metadata,

Column(

"user_id", INT, primary_key=True, test_needs_autoincrement=True

),

Column("user_name", VARCHAR(20)),

Column("extra_data", VARCHAR(20)),

)

metadata.create_all()

@engines.close_first

def teardown(self):

testing.db.execute(users.delete())

@classmethod

def teardown_class(cls):

metadata.drop_all()

def test_cache(self):

conn = testing.db.connect()

cache = {}

cached_conn = conn.execution_options(compiled_cache=cache)

ins = users.insert()

with patch.object(

ins, "compile", Mock(side_effect=ins.compile)

) as compile_mock:

cached_conn.execute(ins, {"user_name": "u1"})

cached_conn.execute(ins, {"user_name": "u2"})

cached_conn.execute(ins, {"user_name": "u3"})

eq_(compile_mock.call_count, 1)

assert len(cache) == 1

eq_(conn.exec_driver_sql("select count(*) from users").scalar(), 3)

@testing.only_on(

["sqlite", "mysql", "postgresql"],

"uses blob value that is problematic for some DBAPIs",

)

@testing.provide_metadata

def test_cache_noleak_on_statement_values(self):

# This is a non regression test for an object reference leak caused

# by the compiled_cache.

metadata = self.metadata

photo = Table(

"photo",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("photo_blob", LargeBinary()),

)

metadata.create_all()

conn = testing.db.connect()

cache = {}

cached_conn = conn.execution_options(compiled_cache=cache)

class PhotoBlob(bytearray):

pass

blob = PhotoBlob(100)

ref_blob = weakref.ref(blob)

ins = photo.insert()

with patch.object(

ins, "compile", Mock(side_effect=ins.compile)

) as compile_mock:

cached_conn.execute(ins, {"photo_blob": blob})

eq_(compile_mock.call_count, 1)

eq_(len(cache), 1)

eq_(conn.exec_driver_sql("select count(*) from photo").scalar(), 1)

del blob

gc_collect()

# The compiled statement cache should not hold any reference to the

# the statement values (only the keys).

eq_(ref_blob(), None)

def test_keys_independent_of_ordering(self):

conn = testing.db.connect()

conn.execute(

users.insert(),

{"user_id": 1, "user_name": "u1", "extra_data": "e1"},

)

cache = {}

cached_conn = conn.execution_options(compiled_cache=cache)

upd = users.update().where(users.c.user_id == bindparam("b_user_id"))

with patch.object(

upd, "compile", Mock(side_effect=upd.compile)

) as compile_mock:

cached_conn.execute(

upd,

util.OrderedDict(

[

("b_user_id", 1),

("user_name", "u2"),

("extra_data", "e2"),

]

),

)

cached_conn.execute(

upd,

util.OrderedDict(

[

("b_user_id", 1),

("extra_data", "e3"),

("user_name", "u3"),

]

),

)

cached_conn.execute(

upd,

util.OrderedDict(

[

("extra_data", "e4"),

("user_name", "u4"),

("b_user_id", 1),

]

),

)

eq_(compile_mock.call_count, 1)

eq_(len(cache), 1)

@testing.requires.schemas

@testing.provide_metadata

def test_schema_translate_in_key(self):

Table("x", self.metadata, Column("q", Integer))

Table(

"x", self.metadata, Column("q", Integer), schema=config.test_schema

)

self.metadata.create_all()

m = MetaData()

t1 = Table("x", m, Column("q", Integer))

ins = t1.insert()

stmt = select([t1.c.q])

cache = {}

with config.db.connect().execution_options(

compiled_cache=cache

) as conn:

conn.execute(ins, {"q": 1})

eq_(conn.scalar(stmt), 1)

with config.db.connect().execution_options(

compiled_cache=cache,

schema_translate_map={None: config.test_schema},

) as conn:

conn.execute(ins, {"q": 2})

eq_(conn.scalar(stmt), 2)

with config.db.connect().execution_options(

compiled_cache=cache, schema_translate_map={None: None},

) as conn:

# should use default schema again even though statement

# was compiled with test_schema in the map

eq_(conn.scalar(stmt), 1)

with config.db.connect().execution_options(

compiled_cache=cache

) as conn:

eq_(conn.scalar(stmt), 1)

class MockStrategyTest(fixtures.TestBase):

def _engine_fixture(self):

buf = util.StringIO()

def dump(sql, *multiparams, **params):

buf.write(util.text_type(sql.compile(dialect=engine.dialect)))

engine = create_mock_engine("postgresql://", executor=dump)

return engine, buf

def test_sequence_not_duped(self):

engine, buf = self._engine_fixture()

metadata = MetaData()

t = Table(