From: Mike Bayer Date: Sun, 30 Sep 2012 05:19:43 +0000 (-0400) Subject: consolidate config into noseplugin, remove the dupe, load noseplugin using imp.load_s... X-Git-Tag: rel_0_8_0b1~102 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=f12813711ad0e0587256a3cf6e3f9f931b9444b8;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git consolidate config into noseplugin, remove the dupe, load noseplugin using imp.load_source(), see if that works --- diff --git a/lib/sqlalchemy/testing/plugin/config.py b/lib/sqlalchemy/testing/plugin/config.py deleted file mode 100644 index 6c92928643..0000000000 --- a/lib/sqlalchemy/testing/plugin/config.py +++ /dev/null @@ -1,196 +0,0 @@ -"""Option and configuration implementations, run by the nose plugin -on test suite startup.""" - -import time -import warnings -import sys -import re - -logging = None -db = None -db_label = None -db_url = None -db_opts = {} -options = None -file_config = 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 _zero_timeout(options, opt_str, value, parser): - warnings.warn("--zero-timeout testing option is now on in all cases") - -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 sys - 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_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 - - # also create alt schemas etc. here? - if options.dropfirst: - e = engines.utf8_engine() - existing = e.table_names() - if existing: - print "Dropping existing tables in database: " + db_url - try: - print "Tables: %s" % ', '.join(existing) - except: - pass - print "Abort within 5 seconds..." - time.sleep(5) - md = schema.MetaData(e, reflect=True) - md.drop_all() - 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 import unitofwork, session, mapper, dependency - from sqlalchemy.util import topological - from sqlalchemy.testing.util import RandomSet - topological.set = unitofwork.set = session.set = mapper.set = \ - dependency.set = RandomSet - -@post -def _requirements(options, file_config): - from sqlalchemy.testing import config - from sqlalchemy import testing - requirement_cls = file_config.get('sqla_testing', "requirement_cls") - - modname, clsname = requirement_cls.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(db, config) - - -@post -def _post_setup_options(opt, file_config): - from sqlalchemy.testing import config - config.options = options - -@post -def _setup_profiling(options, file_config): - from sqlalchemy.testing import profiling - profiling._profile_stats = profiling.ProfileStatsFile( - file_config.get('sqla_testing', 'profile_file')) - diff --git a/lib/sqlalchemy/testing/plugin/noseplugin.py b/lib/sqlalchemy/testing/plugin/noseplugin.py index 7f42f7a68a..cf132198f8 100644 --- a/lib/sqlalchemy/testing/plugin/noseplugin.py +++ b/lib/sqlalchemy/testing/plugin/noseplugin.py @@ -9,16 +9,16 @@ When third party libraries use this plugin, it can be imported normally as "from sqlalchemy.testing.plugin import noseplugin". """ +from __future__ import absolute_import + import os import ConfigParser from nose.plugins import Plugin from nose import SkipTest -from . import config - -from .config import _log, _list_dbs, _zero_timeout, \ - _engine_strategy, _server_side_cursors, pre_configure,\ - post_configure +import time +import sys +import re # late imports fixtures = None @@ -31,6 +31,192 @@ requirements = None util = None file_config = None + +logging = None +db = None +db_label = None +db_url = None +db_opts = {} +options = 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 sys + 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_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 + + # also create alt schemas etc. here? + if options.dropfirst: + e = engines.utf8_engine() + existing = e.table_names() + if existing: + print "Dropping existing tables in database: " + db_url + try: + print "Tables: %s" % ', '.join(existing) + except: + pass + print "Abort within 5 seconds..." + time.sleep(5) + md = schema.MetaData(e, reflect=True) + md.drop_all() + 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 import unitofwork, session, mapper, dependency + from sqlalchemy.util import topological + from sqlalchemy.testing.util import RandomSet + topological.set = unitofwork.set = session.set = mapper.set = \ + dependency.set = RandomSet + +@post +def _requirements(options, file_config): + from sqlalchemy.testing import config + from sqlalchemy import testing + requirement_cls = file_config.get('sqla_testing', "requirement_cls") + + modname, clsname = requirement_cls.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(db, config) + + +@post +def _post_setup_options(opt, file_config): + from sqlalchemy.testing import config + config.options = options + +@post +def _setup_profiling(options, file_config): + from sqlalchemy.testing import profiling + profiling._profile_stats = profiling.ProfileStatsFile( + file_config.get('sqla_testing', 'profile_file')) + + class NoseSQLAlchemy(Plugin): """ Handles the setup and extra properties required for testing SQLAlchemy @@ -59,8 +245,6 @@ class NoseSQLAlchemy(Plugin): 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("--zero-timeout", action="callback", callback=_zero_timeout, - help="Set pool_timeout to zero, applies to QueuePool only") opt("--low-connections", action="store_true", dest="low_connections", help="Use a low number of distinct connections - i.e. for Oracle TNS" ) @@ -90,7 +274,6 @@ class NoseSQLAlchemy(Plugin): global file_config file_config = ConfigParser.ConfigParser() file_config.read(['setup.cfg', 'test.cfg']) - config.file_config = file_config def configure(self, options, conf): Plugin.configure(self, options, conf) @@ -198,5 +381,5 @@ class NoseSQLAlchemy(Plugin): or not issubclass(ctx, fixtures.TestBase): return engines.testing_reaper._stop_test_ctx() - if not config.options.low_connections: + if not options.low_connections: assertions.global_cleanup_assertions() diff --git a/sqla_nose.py b/sqla_nose.py index ec8cc81aee..7865fc978d 100755 --- a/sqla_nose.py +++ b/sqla_nose.py @@ -8,15 +8,18 @@ installs SQLAlchemy's testing plugin into the local environment. """ import sys import os +import imp +import nose from os import path -for pth in ['.', './lib', './test']: - sys.path.insert(0, path.join(path.dirname(path.abspath(__file__)), pth)) +#for pth in ['.', './lib', './test']: +# sys.path.insert(0, path.join(path.dirname(path.abspath(__file__)), pth)) -from plugin.noseplugin import NoseSQLAlchemy -import nose +path = "lib/sqlalchemy/testing/plugin/noseplugin.py" +noseplugin = imp.load_source("noseplugin", path) + -nose.main(addplugins=[NoseSQLAlchemy()]) +nose.main(addplugins=[noseplugin.NoseSQLAlchemy()]) diff --git a/test/plugin/README b/test/plugin/README deleted file mode 100644 index 8abca47e53..0000000000 --- a/test/plugin/README +++ /dev/null @@ -1,36 +0,0 @@ -This is a total hack, as the files here are an exact copy of sqlalchemy/testing/plugin/. - -I can't see much solution here as I'd like "noseplugin" to be available to environments -where SQLAlchemy is installed. However, I also need to be able to load "noseplugin" -*without* importing SQLAlchemy, else coverage doesn't work when testing SQLAlchemy itself. - -A solution here would be if there's a way to get coverage to explicitly include -packages that are already imported. Perhaps like a module reload. - -Other solutions that have not worked include: - - * set an extra, fake sys.path for "sqlalchemy/testing/plugins" and import - relative to that. Breaks in Python 3 as the modules in "sqlalchemy/testing" - get interpreted as Python builtins (i.e. warnings), and apparently ConfigParser - has "import warnings" and it all breaks. - - * Have the noseplugin install as an entirely separate package next to lib/. - We did this in 0.6 and users complained, as we also had it enabled as - a nose plugin. just having it there as a "do-nothing", separately installed - package, is really awkward. Doesn't work. - - - -so: TODO - -1. fix that pdb you have stuck in _do_skips - -2. offload everything in noseplugin that is post-coverage out to -testing/ - mainly _do_skips and the before/after test/context hooks. - -3. merge plugins/config into noseplugin. - -4. remove test/plugin - -5. have sqla_nose.py load noseplugin.py as a single .py file, using -imp or similar. it has no dependencies outside of sqlalchemy.testing. diff --git a/test/plugin/__init__.py b/test/plugin/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/test/plugin/config.py b/test/plugin/config.py deleted file mode 100644 index 6c92928643..0000000000 --- a/test/plugin/config.py +++ /dev/null @@ -1,196 +0,0 @@ -"""Option and configuration implementations, run by the nose plugin -on test suite startup.""" - -import time -import warnings -import sys -import re - -logging = None -db = None -db_label = None -db_url = None -db_opts = {} -options = None -file_config = 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 _zero_timeout(options, opt_str, value, parser): - warnings.warn("--zero-timeout testing option is now on in all cases") - -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 sys - 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_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 - - # also create alt schemas etc. here? - if options.dropfirst: - e = engines.utf8_engine() - existing = e.table_names() - if existing: - print "Dropping existing tables in database: " + db_url - try: - print "Tables: %s" % ', '.join(existing) - except: - pass - print "Abort within 5 seconds..." - time.sleep(5) - md = schema.MetaData(e, reflect=True) - md.drop_all() - 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 import unitofwork, session, mapper, dependency - from sqlalchemy.util import topological - from sqlalchemy.testing.util import RandomSet - topological.set = unitofwork.set = session.set = mapper.set = \ - dependency.set = RandomSet - -@post -def _requirements(options, file_config): - from sqlalchemy.testing import config - from sqlalchemy import testing - requirement_cls = file_config.get('sqla_testing', "requirement_cls") - - modname, clsname = requirement_cls.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(db, config) - - -@post -def _post_setup_options(opt, file_config): - from sqlalchemy.testing import config - config.options = options - -@post -def _setup_profiling(options, file_config): - from sqlalchemy.testing import profiling - profiling._profile_stats = profiling.ProfileStatsFile( - file_config.get('sqla_testing', 'profile_file')) - diff --git a/test/plugin/noseplugin.py b/test/plugin/noseplugin.py deleted file mode 100644 index 7f42f7a68a..0000000000 --- a/test/plugin/noseplugin.py +++ /dev/null @@ -1,202 +0,0 @@ -"""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". - -""" -import os -import ConfigParser - -from nose.plugins import Plugin -from nose import SkipTest -from . import config - -from .config import _log, _list_dbs, _zero_timeout, \ - _engine_strategy, _server_side_cursors, pre_configure,\ - post_configure - -# late imports -fixtures = None -engines = None -exclusions = None -warnings = None -profiling = None -assertions = None -requirements = None -util = None -file_config = None - -class NoseSQLAlchemy(Plugin): - """ - Handles the setup and extra properties required for testing SQLAlchemy - """ - enabled = True - - name = 'sqla_testing' - score = 100 - - 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 (multiple OK)") - opt("--log-debug", action="callback", type="string", callback=_log, - help="turn on debug logging for (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("--zero-timeout", action="callback", callback=_zero_timeout, - help="Set pool_timeout to zero, applies to QueuePool only") - 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("--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']) - config.file_config = file_config - - def configure(self, options, conf): - Plugin.configure(self, options, conf) - self.options = options - for fn in pre_configure: - fn(self.options, file_config) - - def begin(self): - # Lazy setup of other options (post coverage) - for fn in post_configure: - fn(self.options, file_config) - - # late imports, has to happen after config as well - # as nose plugins like coverage - global util, fixtures, engines, exclusions, \ - assertions, warnings, profiling - from sqlalchemy.testing import fixtures, engines, exclusions, \ - assertions, warnings, profiling - from sqlalchemy import util - - def describeTest(self, test): - return "" - - def wantFunction(self, fn): - if fn.__module__.startswith('sqlalchemy.testing'): - 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) - try: - check(test_suite)() - except TypeError: - import pdb - pdb.set_trace() - - if cls.__unsupported_on__: - spec = exclusions.db_spec(*cls.__unsupported_on__) - if spec(config.db): - raise SkipTest( - "'%s' unsupported on DB implementation '%s'" % ( - cls.__name__, config.db.name) - ) - - 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'" % ( - cls.__name__, config.db.name) - ) - - 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))) - - def beforeTest(self, test): - warnings.resetwarnings() - profiling._current_test = test.id() - - def afterTest(self, test): - engines.testing_reaper._after_test_ctx() - warnings.resetwarnings() - - def startContext(self, ctx): - if not isinstance(ctx, type) \ - or not issubclass(ctx, fixtures.TestBase): - return - self._do_skips(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 config.options.low_connections: - assertions.global_cleanup_assertions()