- Support has been added for pytest to run tests. This runner

is currently being supported in addition to nose, and will likely
be preferred to nose going forward.   The nose plugin system used
by SQLAlchemy has been split out so that it works under pytest as
well.  There are no plans to drop support for nose at the moment
and we hope that the test suite itself can continue to remain as
agnostic of testing platform as possible.  See the file
README.unittests.rst for updated information on running tests
with pytest.

The test plugin system has also been enhanced to support running
tests against mutiple database URLs at once, by specifying the ``--db``
and/or ``--dburi`` flags multiple times.  This does not run the entire test
suite for each database, but instead allows test cases that are specific
to certain backends make use of that backend as the test is run.
When using pytest as the test runner, the system will also run
specific test suites multiple times, once for each database, particularly
those tests within the "dialect suite".   The plan is that the enhanced
system will also be used by Alembic, and allow Alembic to run
migration operation tests against multiple backends in one run, including
third-party backends not included within Alembic itself.
Third party dialects and extensions are also encouraged to standardize
on SQLAlchemy's test suite as a basis; see the file README.dialects.rst
for background on building out from SQLAlchemy's test platform.

1 files changed, 31 insertions, 412 deletions

diff --git a/lib/sqlalchemy/testing/plugin/noseplugin.py b/lib/sqlalchemy/testing/plugin/noseplugin.py
index 27a028cd4..a68eb88cf 100644
--- a/lib/sqlalchemy/testing/plugin/noseplugin.py
+++ b/lib/sqlalchemy/testing/plugin/noseplugin.py
@@ -6,276 +6,24 @@
 
 """Enhance nose with extra options and behaviors for running SQLAlchemy tests.
 
-When running ./sqla_nose.py, this module is imported relative to the
-"plugins" package as a top level package by the sqla_nose.py runner,
-so that the plugin can be loaded with the rest of nose including the coverage
-plugin before any of SQLAlchemy itself is imported, so that coverage works.
-
-When third party libraries use this plugin, it can be imported
-normally as "from sqlalchemy.testing.plugin import noseplugin".
+Must be run via ./sqla_nose.py so that it is imported in the expected
+way (e.g. as a package-less import).
 
 """
 
-from __future__ import absolute_import
-
 import os
-import sys
-py3k = sys.version_info >= (3, 0)
-
-if py3k:
-    import configparser
-else:
-    import ConfigParser as configparser
 
 from nose.plugins import Plugin
-from nose import SkipTest
-import sys
-import re
-
-# late imports
 fixtures = None
-engines = None
-exclusions = None
-warnings = None
-profiling = None
-assertions = None
-requirements = None
-config = None
-util = None
-file_config = None
-
-
-logging = None
-db = None
-db_label = None
-db_url = None
-db_opts = {}
-options = None
-_existing_engine = None
-
-
-def _log(option, opt_str, value, parser):
-    global logging
-    if not logging:
-        import logging
-        logging.basicConfig()
-
-    if opt_str.endswith('-info'):
-        logging.getLogger(value).setLevel(logging.INFO)
-    elif opt_str.endswith('-debug'):
-        logging.getLogger(value).setLevel(logging.DEBUG)
-
-
-def _list_dbs(*args):
-    print("Available --db options (use --dburi to override)")
-    for macro in sorted(file_config.options('db')):
-        print("%20s\t%s" % (macro, file_config.get('db', macro)))
-    sys.exit(0)
-
-
-def _server_side_cursors(options, opt_str, value, parser):
-    db_opts['server_side_cursors'] = True
-
-
-def _engine_strategy(options, opt_str, value, parser):
-    if value:
-        db_opts['strategy'] = value
-
-pre_configure = []
-post_configure = []
-
-
-def pre(fn):
-    pre_configure.append(fn)
-    return fn
-
-
-def post(fn):
-    post_configure.append(fn)
-    return fn
-
-
-@pre
-def _setup_options(opt, file_config):
-    global options
-    options = opt
-
-
-@pre
-def _monkeypatch_cdecimal(options, file_config):
-    if options.cdecimal:
-        import cdecimal
-        sys.modules['decimal'] = cdecimal
-
-
-@post
-def _engine_uri(options, file_config):
-    global db_label, db_url
-
-    if options.dburi:
-        db_url = options.dburi
-        db_label = db_url[:db_url.index(':')]
-    elif options.db:
-        db_label = options.db
-        db_url = None
-
-    if db_url is None:
-        if db_label not in file_config.options('db'):
-            raise RuntimeError(
-                "Unknown URI specifier '%s'.  Specify --dbs for known uris."
-                        % db_label)
-        db_url = file_config.get('db', db_label)
-
-
-@post
-def _require(options, file_config):
-    if not(options.require or
-           (file_config.has_section('require') and
-            file_config.items('require'))):
-        return
-
-    try:
-        import pkg_resources
-    except ImportError:
-        raise RuntimeError("setuptools is required for version requirements")
-
-    cmdline = []
-    for requirement in options.require:
-        pkg_resources.require(requirement)
-        cmdline.append(re.split('\s*(<!>=)', requirement, 1)[0])
-
-    if file_config.has_section('require'):
-        for label, requirement in file_config.items('require'):
-            if not label == db_label or label.startswith('%s.' % db_label):
-                continue
-            seen = [c for c in cmdline if requirement.startswith(c)]
-            if seen:
-                continue
-            pkg_resources.require(requirement)
-
-
-@post
-def _engine_pool(options, file_config):
-    if options.mockpool:
-        from sqlalchemy import pool
-        db_opts['poolclass'] = pool.AssertionPool
-
-
-@post
-def _create_testing_engine(options, file_config):
-    from sqlalchemy.testing import engines, config
-    from sqlalchemy import testing
-    global db
-    config.db = testing.db = db = engines.testing_engine(db_url, db_opts)
-    config.db.connect().close()
-    config.db_opts = db_opts
-    config.db_url = db_url
-
-
-@post
-def _prep_testing_database(options, file_config):
-    from sqlalchemy.testing import engines
-    from sqlalchemy import schema, inspect
-
-    # also create alt schemas etc. here?
-    if options.dropfirst:
-        e = engines.utf8_engine()
-        inspector = inspect(e)
 
-        try:
-            view_names = inspector.get_view_names()
-        except NotImplementedError:
-            pass
-        else:
-            for vname in view_names:
-                e.execute(schema._DropView(schema.Table(vname, schema.MetaData())))
-
-        try:
-            view_names = inspector.get_view_names(schema="test_schema")
-        except NotImplementedError:
-            pass
-        else:
-            for vname in view_names:
-                e.execute(schema._DropView(
-                            schema.Table(vname,
-                                        schema.MetaData(), schema="test_schema")))
-
-        for tname in reversed(inspector.get_table_names(order_by="foreign_key")):
-            e.execute(schema.DropTable(schema.Table(tname, schema.MetaData())))
-
-        for tname in reversed(inspector.get_table_names(
-                                order_by="foreign_key", schema="test_schema")):
-            e.execute(schema.DropTable(
-                schema.Table(tname, schema.MetaData(), schema="test_schema")))
-
-        e.dispose()
-
-
-@post
-def _set_table_options(options, file_config):
-    from sqlalchemy.testing import schema
-
-    table_options = schema.table_options
-    for spec in options.tableopts:
-        key, value = spec.split('=')
-        table_options[key] = value
-
-    if options.mysql_engine:
-        table_options['mysql_engine'] = options.mysql_engine
-
-
-@post
-def _reverse_topological(options, file_config):
-    if options.reversetop:
-        from sqlalchemy.orm.util import randomize_unitofwork
-        randomize_unitofwork()
-
-
-def _requirements_opt(options, opt_str, value, parser):
-    _setup_requirements(value)
-
-@post
-def _requirements(options, file_config):
-
-    requirement_cls = file_config.get('sqla_testing', "requirement_cls")
-    _setup_requirements(requirement_cls)
-
-def _setup_requirements(argument):
-    from sqlalchemy.testing import config
-    from sqlalchemy import testing
-
-    if config.requirements is not None:
-        return
-
-    modname, clsname = argument.split(":")
-
-    # importlib.import_module() only introduced in 2.7, a little
-    # late
-    mod = __import__(modname)
-    for component in modname.split(".")[1:]:
-        mod = getattr(mod, component)
-    req_cls = getattr(mod, clsname)
-    config.requirements = testing.requires = req_cls(config)
-
-
-@post
-def _post_setup_options(opt, file_config):
-    from sqlalchemy.testing import config
-    config.options = options
-    config.file_config = file_config
-
-
-@post
-def _setup_profiling(options, file_config):
-    from sqlalchemy.testing import profiling
-    profiling._profile_stats = profiling.ProfileStatsFile(
-                file_config.get('sqla_testing', 'profile_file'))
+# no package imports yet!  this prevents us from tripping coverage
+# too soon.
+import imp
+path = os.path.join(os.path.dirname(__file__), "plugin_base.py")
+plugin_base = imp.load_source("plugin_base", path)
 
 
 class NoseSQLAlchemy(Plugin):
-    """
-    Handles the setup and extra properties required for testing SQLAlchemy
-    """
     enabled = True
 
     name = 'sqla_testing'
@@ -284,74 +32,29 @@ class NoseSQLAlchemy(Plugin):
     def options(self, parser, env=os.environ):
         Plugin.options(self, parser, env)
         opt = parser.add_option
-        opt("--log-info", action="callback", type="string", callback=_log,
-            help="turn on info logging for <LOG> (multiple OK)")
-        opt("--log-debug", action="callback", type="string", callback=_log,
-            help="turn on debug logging for <LOG> (multiple OK)")
-        opt("--require", action="append", dest="require", default=[],
-            help="require a particular driver or module version (multiple OK)")
-        opt("--db", action="store", dest="db", default="default",
-            help="Use prefab database uri")
-        opt('--dbs', action='callback', callback=_list_dbs,
-            help="List available prefab dbs")
-        opt("--dburi", action="store", dest="dburi",
-            help="Database uri (overrides --db)")
-        opt("--dropfirst", action="store_true", dest="dropfirst",
-            help="Drop all tables in the target database first")
-        opt("--mockpool", action="store_true", dest="mockpool",
-            help="Use mock pool (asserts only one connection used)")
-        opt("--low-connections", action="store_true", dest="low_connections",
-            help="Use a low number of distinct connections - i.e. for Oracle TNS"
-        )
-        opt("--enginestrategy", action="callback", type="string",
-            callback=_engine_strategy,
-            help="Engine strategy (plain or threadlocal, defaults to plain)")
-        opt("--reversetop", action="store_true", dest="reversetop", default=False,
-            help="Use a random-ordering set implementation in the ORM (helps "
-                  "reveal dependency issues)")
-        opt("--requirements", action="callback", type="string",
-            callback=_requirements_opt,
-            help="requirements class for testing, overrides setup.cfg")
-        opt("--with-cdecimal", action="store_true", dest="cdecimal", default=False,
-            help="Monkeypatch the cdecimal library into Python 'decimal' for all tests")
-        opt("--unhashable", action="store_true", dest="unhashable", default=False,
-            help="Disallow SQLAlchemy from performing a hash() on mapped test objects.")
-        opt("--noncomparable", action="store_true", dest="noncomparable", default=False,
-            help="Disallow SQLAlchemy from performing == on mapped test objects.")
-        opt("--truthless", action="store_true", dest="truthless", default=False,
-            help="Disallow SQLAlchemy from truth-evaluating mapped test objects.")
-        opt("--serverside", action="callback", callback=_server_side_cursors,
-            help="Turn on server side cursors for PG")
-        opt("--mysql-engine", action="store", dest="mysql_engine", default=None,
-            help="Use the specified MySQL storage engine for all tables, default is "
-                 "a db-default/InnoDB combo.")
-        opt("--table-option", action="append", dest="tableopts", default=[],
-            help="Add a dialect-specific table option, key=value")
-        opt("--write-profiles", action="store_true", dest="write_profiles", default=False,
-                help="Write/update profiling data.")
-        global file_config
-        file_config = configparser.ConfigParser()
-        file_config.read(['setup.cfg', 'test.cfg'])
+
+        def make_option(name, **kw):
+            callback_ = kw.pop("callback", None)
+            if callback_:
+                def wrap_(option, opt_str, value, parser):
+                    callback_(opt_str, value, parser)
+                kw["callback"] = wrap_
+            opt(name, **kw)
+
+        plugin_base.setup_options(make_option)
+        plugin_base.read_config()
 
     def configure(self, options, conf):
-        Plugin.configure(self, options, conf)
-        self.options = options
-        for fn in pre_configure:
-            fn(self.options, file_config)
+        super(NoseSQLAlchemy, self).configure(options, conf)
+        plugin_base.pre_begin(options)
 
-    def begin(self):
-        # Lazy setup of other options (post coverage)
-        for fn in post_configure:
-            fn(self.options, file_config)
+        plugin_base.set_coverage_flag(options.enable_plugin_coverage)
 
-        # late imports, has to happen after config as well
-        # as nose plugins like coverage
-        global util, fixtures, engines, exclusions, \
-                        assertions, warnings, profiling,\
-                        config
-        from sqlalchemy.testing import fixtures, engines, exclusions, \
-                        assertions, warnings, profiling, config
-        from sqlalchemy import util
+        global fixtures
+        from sqlalchemy.testing import fixtures
+
+    def begin(self):
+        plugin_base.post_begin()
 
     def describeTest(self, test):
         return ""
@@ -363,106 +66,22 @@ class NoseSQLAlchemy(Plugin):
             return False
 
     def wantClass(self, cls):
-        """Return true if you want the main test selector to collect
-        tests from this class, false if you don't, and None if you don't
-        care.
-
-        :Parameters:
-           cls : class
-             The class being examined by the selector
-
-        """
-        if not issubclass(cls, fixtures.TestBase):
-            return False
-        elif cls.__name__.startswith('_'):
-            return False
-        else:
-            return True
-
-    def _do_skips(self, cls):
-        from sqlalchemy.testing import config
-        if hasattr(cls, '__requires__'):
-            def test_suite():
-                return 'ok'
-            test_suite.__name__ = cls.__name__
-            for requirement in cls.__requires__:
-                check = getattr(config.requirements, requirement)
-
-                if not check.enabled:
-                    raise SkipTest(
-                        check.reason if check.reason
-                        else
-                        (
-                            "'%s' unsupported on DB implementation '%s' == %s" % (
-                                cls.__name__, config.db.name,
-                                config.db.dialect.server_version_info
-                            )
-                        )
-                    )
-
-        if cls.__unsupported_on__:
-            spec = exclusions.db_spec(*cls.__unsupported_on__)
-            if spec(config.db):
-                raise SkipTest(
-                    "'%s' unsupported on DB implementation '%s' == %s" % (
-                     cls.__name__, config.db.name,
-                        config.db.dialect.server_version_info)
-                    )
-
-        if getattr(cls, '__only_on__', None):
-            spec = exclusions.db_spec(*util.to_list(cls.__only_on__))
-            if not spec(config.db):
-                raise SkipTest(
-                    "'%s' unsupported on DB implementation '%s' == %s" % (
-                     cls.__name__, config.db.name,
-                        config.db.dialect.server_version_info)
-                    )
-
-        if getattr(cls, '__skip_if__', False):
-            for c in getattr(cls, '__skip_if__'):
-                if c():
-                    raise SkipTest("'%s' skipped by %s" % (
-                        cls.__name__, c.__name__)
-                    )
-
-        for db, op, spec in getattr(cls, '__excluded_on__', ()):
-            exclusions.exclude(db, op, spec,
-                    "'%s' unsupported on DB %s version %s" % (
-                    cls.__name__, config.db.name,
-                    exclusions._server_version(config.db)))
+        return plugin_base.want_class(cls)
 
     def beforeTest(self, test):
-        warnings.resetwarnings()
-        profiling._current_test = test.id()
+        plugin_base.before_test(test, test.id())
 
     def afterTest(self, test):
-        engines.testing_reaper._after_test_ctx()
-        warnings.resetwarnings()
-
-    def _setup_engine(self, ctx):
-        if getattr(ctx, '__engine_options__', None):
-            global _existing_engine
-            _existing_engine = config.db
-            config.db = engines.testing_engine(options=ctx.__engine_options__)
-
-    def _restore_engine(self, ctx):
-        global _existing_engine
-        if _existing_engine is not None:
-            config.db = _existing_engine
-            _existing_engine = None
+        plugin_base.after_test(test)
 
     def startContext(self, ctx):
         if not isinstance(ctx, type) \
             or not issubclass(ctx, fixtures.TestBase):
             return
-        self._do_skips(ctx)
-        self._setup_engine(ctx)
+        plugin_base.start_test_class(ctx)
 
     def stopContext(self, ctx):
         if not isinstance(ctx, type) \
             or not issubclass(ctx, fixtures.TestBase):
             return
-        engines.testing_reaper._stop_test_ctx()
-        if not options.low_connections:
-            assertions.global_cleanup_assertions()
-        self._restore_engine(ctx)
+        plugin_base.stop_test_class(ctx)