Commits

Mike Bayer committed dd866a3

- fixes for mxODBC, some pyodbc
- enhancements to test suite including ability to set up a testing engine
for a whole test class, fixes to how noseplugin sets up/tears
down per-class context

Comments (0)

Files changed (13)

lib/sqlalchemy/connectors/mxodbc.py

     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))

lib/sqlalchemy/dialects/mssql/base.py

             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

lib/sqlalchemy/dialects/mssql/mxodbc.py

 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
 

lib/sqlalchemy/dialects/mssql/pyodbc.py

         }
     )
 
-    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 \

lib/sqlalchemy/engine/base.py

                 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

test/bootstrap/noseplugin.py

         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()

test/engine/test_execute.py

 
         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', {}, ()),

test/lib/fixtures.py

     # 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
 

test/lib/requires.py

         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',

test/sql/test_defaults.py

 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):

test/sql/test_query.py

                               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(

test/sql/test_types.py

         """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

test/sql/test_update.py

         )
 
 class UpdateFromRoundTripTest(_UpdateFromTestBase, fixtures.TablesTest):
+    __testing_engine__ = {'execution_options':{'native_odbc_execute':False}}
 
     @testing.requires.update_from
     def test_exec_two_table(self):