driver='mxodbc'
supports_sane_multi_rowcount = False
- supports_unicode_statements = False
- supports_unicode_binds = False
+ supports_unicode_statements = True
+ supports_unicode_binds = True
supports_native_decimal = True
version.append(n)
return tuple(version)
- def do_execute(self, cursor, statement, parameters, context=None):
+ def _get_direct(self, context):
+ return True
if context:
native_odbc_execute = context.execution_options.\
get('native_odbc_execute', 'auto')
if native_odbc_execute is True:
# user specified native_odbc_execute=True
- cursor.execute(statement, parameters)
+ return False
elif native_odbc_execute is False:
# user specified native_odbc_execute=False
- cursor.executedirect(statement, parameters)
+ return True
elif context.is_crud:
# statement is UPDATE, DELETE, INSERT
- cursor.execute(statement, parameters)
+ return False
else:
# all other statements
- cursor.executedirect(statement, parameters)
+ return True
else:
- cursor.executedirect(statement, parameters)
+ return True
+
+ def do_executemany(self, cursor, statement, parameters, context=None):
+ cursor.executemany(statement, parameters, direct=self._get_direct(context))
+
+ def do_execute(self, cursor, statement, parameters, context=None):
+ cursor.execute(statement, parameters, direct=self._get_direct(context))
else:
return value
return process
+_MSTime = TIME
class _DateTimeBase(object):
def bind_processor(self, dialect):
self.process(binary.right, **kw)
)
- def visit_function(self, func, **kw):
- kw['literal_binds'] = True
- return super(MSSQLStrictCompiler, self).visit_function(func, **kw)
-
def render_literal_value(self, value, type_):
"""
For date and datetime values, convert to a string
from .pyodbc import MSExecutionContext_pyodbc
from .base import (MSDialect,
MSSQLStrictCompiler,
- _MSDateTime, _MSDate, TIME)
+ _MSDateTime, _MSDate, _MSTime)
+class _MSDate_mxodbc(_MSDate):
+ def bind_processor(self, dialect):
+ def process(value):
+ if value is not None:
+ return "%s-%s-%s" % (value.year, value.month, value.day)
+ else:
+ return None
+ return process
+
+class _MSTime_mxodbc(_MSTime):
+ def bind_processor(self, dialect):
+ def process(value):
+ if value is not None:
+ return "%s:%s:%s" % (value.hour, value.minute, value.second)
+ else:
+ return None
+ return process
+
class MSExecutionContext_mxodbc(MSExecutionContext_pyodbc):
"""
The pyodbc execution context is useful for enabling
colspecs = {
#sqltypes.Numeric : _MSNumeric,
sqltypes.DateTime : _MSDateTime,
- sqltypes.Date : _MSDate,
- sqltypes.Time : TIME,
+ sqltypes.Date : _MSDate_mxodbc,
+ sqltypes.Time : _MSTime_mxodbc,
}
- def __init__(self, description_encoding='latin-1', **params):
+ def __init__(self, description_encoding=None, **params):
super(MSDialect_mxodbc, self).__init__(**params)
self.description_encoding = description_encoding
}
)
- def __init__(self, description_encoding='latin-1', **params):
+ def __init__(self, description_encoding=None, **params):
super(MSDialect_pyodbc, self).__init__(**params)
self.description_encoding = description_encoding
self.use_scope_identity = self.use_scope_identity and \
ex_text = str(e)
except TypeError:
ex_text = repr(e)
- self.connection._logger.warn("Error closing cursor: %s", ex_text)
+ if not self.closed:
+ self.connection._logger.warn(
+ "Error closing cursor: %s", ex_text)
if isinstance(e, (SystemExit, KeyboardInterrupt)):
raise
elif cls.__name__.startswith('_'):
return False
else:
- if hasattr(cls, 'setup_class'):
- existing_setup = cls.setup_class.im_func
- else:
- existing_setup = None
- @classmethod
- def setup_class(cls):
- self._do_skips(cls)
- if existing_setup:
- existing_setup(cls)
- cls.setup_class = setup_class
-
return True
def _do_skips(self, cls):
)
for db, op, spec in getattr(cls, '__excluded_on__', ()):
- testing.exclude(db, op, spec, "'%s' unsupported on DB %s version %s" % (
+ testing.exclude(db, op, spec,
+ "'%s' unsupported on DB %s version %s" % (
cls.__name__, testing.db.name,
testing._server_version()))
engines.testing_reaper._after_test_ctx()
testing.resetwarnings()
+ def _setup_cls_engines(self, cls):
+ engine_opts = getattr(cls, '__testing_engine__', None)
+ if engine_opts:
+ self._save_testing_db = testing.db
+ testing.db = engines.testing_engine(options=engine_opts)
+
+ def _teardown_cls_engines(self, cls):
+ engine_opts = getattr(cls, '__testing_engine__', None)
+ if engine_opts:
+ testing.db = self._save_testing_db
+ del self._save_testing_db
+
+ def startContext(self, ctx):
+ if not isinstance(ctx, type) \
+ or not issubclass(ctx, fixtures.TestBase):
+ return
+ self._do_skips(ctx)
+ self._setup_cls_engines(ctx)
+
def stopContext(self, ctx):
+ if not isinstance(ctx, type) \
+ or not issubclass(ctx, fixtures.TestBase):
+ return
engines.testing_reaper._stop_test_ctx()
+ self._teardown_cls_engines(ctx)
if not config.options.low_connections:
testing.global_cleanup_assertions()
for engine in [
engines.testing_engine(options=dict(implicit_returning=False)),
- #engines.testing_engine(options=dict(implicit_returning=False,
- # strategy='threadlocal')),
- #engines.testing_engine(options=dict(implicit_returning=False)).\
- # connect()
+ engines.testing_engine(options=dict(implicit_returning=False,
+ strategy='threadlocal')),
+ engines.testing_engine(options=dict(implicit_returning=False)).\
+ connect()
]:
event.listen(engine, 'before_execute', execute)
event.listen(engine, 'before_cursor_execute', cursor_execute)
('INSERT INTO t1 (c1, c2)', {
'c2': 'some data', 'c1': 5},
(5, 'some data')),
- ('SELECT lower', {'lower_2': 'Foo'}, ('Foo', )),
+ ('SELECT lower', {'lower_2': 'Foo'},
+ () if testing.against('mssql+mxodbc') else
+ ('Foo', )),
('INSERT INTO t1 (c1, c2)',
{'c2': 'foo', 'c1': 6},
(6, 'foo')),
('CREATE TABLE t1', {}, ()),
('INSERT INTO t1 (c1, c2)', {'c2': 'some data', 'c1'
: 5}, (5, 'some data')),
- ('SELECT lower', {'lower_2': 'Foo'}, ('Foo', )),
+ ('SELECT lower', {'lower_2': 'Foo'},
+ () if testing.against('mssql+mxodbc')
+ else ('Foo', )),
('INSERT INTO t1 (c1, c2)', {'c2': 'foo', 'c1': 6},
(6, 'foo')),
('select * from t1', {}, ()),
# skipped.
__skip_if__ = None
+ # replace testing.db with a testing.engine()
+ # for the duration of this suite, using the given
+ # arguments
+ __testing_engine__ = None
+
def assert_(self, val, msg=None):
assert val, msg
no_support('informix', 'not supported by database'),
)
+def standalone_binds(fn):
+ """target database/driver supports bound parameters as column expressions
+ without being in the context of a typed column.
+
+ """
+ return _chain_decorators_on(
+ fn,
+ no_support('firebird', 'not supported by driver'),
+ no_support('mssql+mxodbc', 'not supported by driver')
+ )
+
def identity(fn):
"""Target database must support GENERATED AS IDENTITY or a facsimile.
no_support('sybase', 'not supported by database'),
)
+def binary_comparisons(fn):
+ """target database/driver can allow BLOB/BINARY fields to be compared
+ against a bound parameter value.
+ """
+ return _chain_decorators_on(
+ fn,
+ no_support('oracle', 'not supported by database/driver'),
+ no_support('mssql', 'not supported by database/driver')
+ )
+
def independent_cursors(fn):
"""Target must support simultaneous, independent database cursors on a single connection."""
)
+def emulated_lastrowid(fn):
+ """"target dialect retrieves cursor.lastrowid or an equivalent
+ after an insert() construct executes.
+ """
+ return _chain_decorators_on(
+ fn,
+ fails_on_everything_except('mysql+mysqldb', 'mysql+oursql',
+ 'sqlite+pysqlite', 'mysql+pymysql',
+ 'mssql+pyodbc', 'mssql+mxodbc'),
+ )
+
def dbapi_lastrowid(fn):
+ """"target backend includes a 'lastrowid' accessor on the DBAPI
+ cursor object.
+ """
return _chain_decorators_on(
fn,
fails_on_everything_except('mysql+mysqldb', 'mysql+oursql',
from test.lib import fixtures
class DefaultTest(fixtures.TestBase):
+ __testing_engine__ = {'execution_options':{'native_odbc_execute':False}}
@classmethod
def setup_class(cls):
class PKDefaultTest(fixtures.TablesTest):
__requires__ = ('subqueries',)
+ __testing_engine__ = {'execution_options':{'native_odbc_execute':False}}
@classmethod
def define_tables(cls, metadata):
class PKIncrementTest(fixtures.TablesTest):
run_define_tables = 'each'
+ __testing_engine__ = {'execution_options':{'native_odbc_execute':False}}
@classmethod
def define_tables(cls, metadata):
use_labels=labels),
[(3, 'a'), (2, 'b'), (1, None)])
+ @testing.fails_on('mssql+pyodbc',
+ "pyodbc result row doesn't support slicing")
def test_column_slices(self):
users.insert().execute(user_id=1, user_name='john')
users.insert().execute(user_id=2, user_name='jack')
stmt, {'data': 'data'}
)
+ @testing.requires.standalone_binds
def test_select_columns(self):
stmt = select([bindparam('data'), bindparam('x')])
self._assert_raises(
inserted_primary_key=[1]
)
+ def test_uppercase_direct_params(self):
+ t = self.tables.foo
+ self._test(
+ t.insert().values(id=1, data='data', x=5),
+ (1, 'data', 5),
+ inserted_primary_key=[1]
+ )
+
+ @testing.requires.returning
+ def test_uppercase_direct_params_returning(self):
+ t = self.tables.foo
+ self._test(
+ t.insert().values(
+ id=1, data='data', x=5).returning(t.c.id, t.c.x),
+ (1, 'data', 5),
+ returning=(1, 5)
+ )
+
+ @testing.fails_on('mssql',
+ "lowercase table doesn't support identity insert disable")
def test_direct_params(self):
t = self._fixture()
self._test(
inserted_primary_key=[]
)
+ @testing.fails_on('mssql',
+ "lowercase table doesn't support identity insert disable")
@testing.requires.returning
def test_direct_params_returning(self):
t = self._fixture()
returning=(1, 5)
)
- @testing.requires.dbapi_lastrowid
+ @testing.requires.emulated_lastrowid
def test_implicit_pk(self):
t = self._fixture()
self._test(
inserted_primary_key=[]
)
- @testing.requires.dbapi_lastrowid
+ @testing.requires.emulated_lastrowid
def test_implicit_pk_multi_rows(self):
t = self._fixture()
self._test_multi(
],
)
- @testing.requires.dbapi_lastrowid
+ @testing.requires.emulated_lastrowid
def test_implicit_pk_inline(self):
t = self._fixture()
self._test(
"""assert expected values for 'native unicode' mode"""
if \
- (testing.against('mssql+pyodbc') and not testing.db.dialect.freetds):
+ (testing.against('mssql+pyodbc') and not testing.db.dialect.freetds) \
+ or testing.against('mssql+mxodbc'):
assert testing.db.dialect.returns_unicode_strings == 'conditional'
return
# lambda: testing.db_spec("postgresql")(testing.db),
# "pg8000 and psycopg2 both have issues here in py3k"
# )
+ @testing.skip_if(lambda: testing.db_spec('mssql+mxodbc'),
+ "unsupported behavior")
def test_ignoring_unicode_error(self):
"""checks String(unicode_error='ignore') is passed to underlying codec."""
class BinaryTest(fixtures.TestBase, AssertsExecutionResults):
__excluded_on__ = (
('mysql', '<', (4, 1, 1)), # screwy varbinary types
- )
-
+ )
+
@classmethod
def setup_class(cls):
global binary_table, MyPickleType, metadata
eq_(testobj3.moredata, l[0]['mypickle'].moredata)
eq_(l[0]['mypickle'].stuff, 'this is the right stuff')
- @testing.fails_on('oracle+cx_oracle', 'oracle fairly grumpy about binary '
- 'data, not really known how to make this work')
+ @testing.requires.binary_comparisons
def test_comparison(self):
"""test that type coercion occurs on comparison for binary"""
return value / 10
return process
def adapt_operator(self, op):
- return {operators.add:operators.sub, operators.sub:operators.add}.get(op, op)
+ return {operators.add:operators.sub,
+ operators.sub:operators.add}.get(op, op)
class MyTypeDec(types.TypeDecorator):
impl = String
)
class UpdateFromRoundTripTest(_UpdateFromTestBase, fixtures.TablesTest):
+ __testing_engine__ = {'execution_options':{'native_odbc_execute':False}}
@testing.requires.update_from
def test_exec_two_table(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
