1
"""Enhance nose with extra options and behaviors for running SQLAlchemy tests.
3
When running ./sqla_nose.py, this module is imported relative to the
4
"plugins" package as a top level package by the sqla_nose.py runner,
5
so that the plugin can be loaded with the rest of nose including the coverage
6
plugin before any of SQLAlchemy itself is imported, so that coverage works.
8
When third party libraries use this plugin, it can be imported
9
normally as "from sqlalchemy.testing.plugin import noseplugin".
12
from __future__ import absolute_import
17
from nose.plugins import Plugin
18
from nose import SkipTest
42
_existing_engine = None
45
def _log(option, opt_str, value, parser):
51
if opt_str.endswith('-info'):
52
logging.getLogger(value).setLevel(logging.INFO)
53
elif opt_str.endswith('-debug'):
54
logging.getLogger(value).setLevel(logging.DEBUG)
58
print "Available --db options (use --dburi to override)"
59
for macro in sorted(file_config.options('db')):
60
print "%20s\t%s" % (macro, file_config.get('db', macro))
64
def _server_side_cursors(options, opt_str, value, parser):
65
db_opts['server_side_cursors'] = True
68
def _engine_strategy(options, opt_str, value, parser):
70
db_opts['strategy'] = value
77
pre_configure.append(fn)
82
post_configure.append(fn)
87
def _setup_options(opt, file_config):
93
def _monkeypatch_cdecimal(options, file_config):
96
sys.modules['decimal'] = cdecimal
100
def _engine_uri(options, file_config):
101
global db_label, db_url
104
db_url = options.dburi
105
db_label = db_url[:db_url.index(':')]
107
db_label = options.db
111
if db_label not in file_config.options('db'):
113
"Unknown URI specifier '%s'. Specify --dbs for known uris."
115
db_url = file_config.get('db', db_label)
119
def _require(options, file_config):
120
if not(options.require or
121
(file_config.has_section('require') and
122
file_config.items('require'))):
128
raise RuntimeError("setuptools is required for version requirements")
131
for requirement in options.require:
132
pkg_resources.require(requirement)
133
cmdline.append(re.split('\s*(<!>=)', requirement, 1)[0])
135
if file_config.has_section('require'):
136
for label, requirement in file_config.items('require'):
137
if not label == db_label or label.startswith('%s.' % db_label):
139
seen = [c for c in cmdline if requirement.startswith(c)]
142
pkg_resources.require(requirement)
146
def _engine_pool(options, file_config):
148
from sqlalchemy import pool
149
db_opts['poolclass'] = pool.AssertionPool
153
def _create_testing_engine(options, file_config):
154
from sqlalchemy.testing import engines, config
155
from sqlalchemy import testing
157
config.db = testing.db = db = engines.testing_engine(db_url, db_opts)
158
config.db.connect().close()
159
config.db_opts = db_opts
160
config.db_url = db_url
164
def _prep_testing_database(options, file_config):
165
from sqlalchemy.testing import engines
166
from sqlalchemy import schema, inspect
168
# also create alt schemas etc. here?
169
if options.dropfirst:
170
e = engines.utf8_engine()
171
inspector = inspect(e)
174
view_names = inspector.get_view_names()
175
except NotImplementedError:
178
for vname in view_names:
179
e.execute(schema._DropView(schema.Table(vname, schema.MetaData())))
182
view_names = inspector.get_view_names(schema="test_schema")
183
except NotImplementedError:
186
for vname in view_names:
187
e.execute(schema._DropView(
189
schema.MetaData(), schema="test_schema")))
191
for tname in reversed(inspector.get_table_names(order_by="foreign_key")):
192
e.execute(schema.DropTable(schema.Table(tname, schema.MetaData())))
194
for tname in reversed(inspector.get_table_names(
195
order_by="foreign_key", schema="test_schema")):
196
e.execute(schema.DropTable(
197
schema.Table(tname, schema.MetaData(), schema="test_schema")))
203
def _set_table_options(options, file_config):
204
from sqlalchemy.testing import schema
206
table_options = schema.table_options
207
for spec in options.tableopts:
208
key, value = spec.split('=')
209
table_options[key] = value
211
if options.mysql_engine:
212
table_options['mysql_engine'] = options.mysql_engine
216
def _reverse_topological(options, file_config):
217
if options.reversetop:
218
from sqlalchemy.orm.util import randomize_unitofwork
219
randomize_unitofwork()
222
def _requirements_opt(options, opt_str, value, parser):
223
_setup_requirements(value)
226
def _requirements(options, file_config):
228
requirement_cls = file_config.get('sqla_testing', "requirement_cls")
229
_setup_requirements(requirement_cls)
231
def _setup_requirements(argument):
232
from sqlalchemy.testing import config
233
from sqlalchemy import testing
235
if config.requirements is not None:
238
modname, clsname = argument.split(":")
240
# importlib.import_module() only introduced in 2.7, a little
242
mod = __import__(modname)
243
for component in modname.split(".")[1:]:
244
mod = getattr(mod, component)
245
req_cls = getattr(mod, clsname)
246
config.requirements = testing.requires = req_cls(config)
250
def _post_setup_options(opt, file_config):
251
from sqlalchemy.testing import config
252
config.options = options
253
config.file_config = file_config
257
def _setup_profiling(options, file_config):
258
from sqlalchemy.testing import profiling
259
profiling._profile_stats = profiling.ProfileStatsFile(
260
file_config.get('sqla_testing', 'profile_file'))
263
class NoseSQLAlchemy(Plugin):
265
Handles the setup and extra properties required for testing SQLAlchemy
269
name = 'sqla_testing'
272
def options(self, parser, env=os.environ):
273
Plugin.options(self, parser, env)
274
opt = parser.add_option
275
opt("--log-info", action="callback", type="string", callback=_log,
276
help="turn on info logging for <LOG> (multiple OK)")
277
opt("--log-debug", action="callback", type="string", callback=_log,
278
help="turn on debug logging for <LOG> (multiple OK)")
279
opt("--require", action="append", dest="require", default=[],
280
help="require a particular driver or module version (multiple OK)")
281
opt("--db", action="store", dest="db", default="default",
282
help="Use prefab database uri")
283
opt('--dbs', action='callback', callback=_list_dbs,
284
help="List available prefab dbs")
285
opt("--dburi", action="store", dest="dburi",
286
help="Database uri (overrides --db)")
287
opt("--dropfirst", action="store_true", dest="dropfirst",
288
help="Drop all tables in the target database first")
289
opt("--mockpool", action="store_true", dest="mockpool",
290
help="Use mock pool (asserts only one connection used)")
291
opt("--low-connections", action="store_true", dest="low_connections",
292
help="Use a low number of distinct connections - i.e. for Oracle TNS"
294
opt("--enginestrategy", action="callback", type="string",
295
callback=_engine_strategy,
296
help="Engine strategy (plain or threadlocal, defaults to plain)")
297
opt("--reversetop", action="store_true", dest="reversetop", default=False,
298
help="Use a random-ordering set implementation in the ORM (helps "
299
"reveal dependency issues)")
300
opt("--requirements", action="callback", type="string",
301
callback=_requirements_opt,
302
help="requirements class for testing, overrides setup.cfg")
303
opt("--with-cdecimal", action="store_true", dest="cdecimal", default=False,
304
help="Monkeypatch the cdecimal library into Python 'decimal' for all tests")
305
opt("--unhashable", action="store_true", dest="unhashable", default=False,
306
help="Disallow SQLAlchemy from performing a hash() on mapped test objects.")
307
opt("--noncomparable", action="store_true", dest="noncomparable", default=False,
308
help="Disallow SQLAlchemy from performing == on mapped test objects.")
309
opt("--truthless", action="store_true", dest="truthless", default=False,
310
help="Disallow SQLAlchemy from truth-evaluating mapped test objects.")
311
opt("--serverside", action="callback", callback=_server_side_cursors,
312
help="Turn on server side cursors for PG")
313
opt("--mysql-engine", action="store", dest="mysql_engine", default=None,
314
help="Use the specified MySQL storage engine for all tables, default is "
315
"a db-default/InnoDB combo.")
316
opt("--table-option", action="append", dest="tableopts", default=[],
317
help="Add a dialect-specific table option, key=value")
318
opt("--write-profiles", action="store_true", dest="write_profiles", default=False,
319
help="Write/update profiling data.")
321
file_config = ConfigParser.ConfigParser()
322
file_config.read(['setup.cfg', 'test.cfg'])
324
def configure(self, options, conf):
325
Plugin.configure(self, options, conf)
326
self.options = options
327
for fn in pre_configure:
328
fn(self.options, file_config)
331
# Lazy setup of other options (post coverage)
332
for fn in post_configure:
333
fn(self.options, file_config)
335
# late imports, has to happen after config as well
336
# as nose plugins like coverage
337
global util, fixtures, engines, exclusions, \
338
assertions, warnings, profiling,\
340
from sqlalchemy.testing import fixtures, engines, exclusions, \
341
assertions, warnings, profiling, config
342
from sqlalchemy import util
344
def describeTest(self, test):
347
def wantFunction(self, fn):
348
if fn.__module__.startswith('sqlalchemy.testing'):
351
def wantClass(self, cls):
352
"""Return true if you want the main test selector to collect
353
tests from this class, false if you don't, and None if you don't
358
The class being examined by the selector
361
if not issubclass(cls, fixtures.TestBase):
363
elif cls.__name__.startswith('_'):
368
def _do_skips(self, cls):
369
from sqlalchemy.testing import config
370
if hasattr(cls, '__requires__'):
373
test_suite.__name__ = cls.__name__
374
for requirement in cls.__requires__:
375
check = getattr(config.requirements, requirement)
377
if not check.enabled:
379
check.reason if check.reason
382
"'%s' unsupported on DB implementation '%s'" % (
383
cls.__name__, config.db.name
388
if cls.__unsupported_on__:
389
spec = exclusions.db_spec(*cls.__unsupported_on__)
392
"'%s' unsupported on DB implementation '%s'" % (
393
cls.__name__, config.db.name)
396
if getattr(cls, '__only_on__', None):
397
spec = exclusions.db_spec(*util.to_list(cls.__only_on__))
398
if not spec(config.db):
400
"'%s' unsupported on DB implementation '%s'" % (
401
cls.__name__, config.db.name)
404
if getattr(cls, '__skip_if__', False):
405
for c in getattr(cls, '__skip_if__'):
407
raise SkipTest("'%s' skipped by %s" % (
408
cls.__name__, c.__name__)
411
for db, op, spec in getattr(cls, '__excluded_on__', ()):
412
exclusions.exclude(db, op, spec,
413
"'%s' unsupported on DB %s version %s" % (
414
cls.__name__, config.db.name,
415
exclusions._server_version(config.db)))
417
def beforeTest(self, test):
418
warnings.resetwarnings()
419
profiling._current_test = test.id()
421
def afterTest(self, test):
422
engines.testing_reaper._after_test_ctx()
423
warnings.resetwarnings()
425
def _setup_engine(self, ctx):
426
if getattr(ctx, '__engine_options__', None):
427
global _existing_engine
428
_existing_engine = config.db
429
config.db = engines.testing_engine(options=ctx.__engine_options__)
431
def _restore_engine(self, ctx):
432
global _existing_engine
433
if _existing_engine is not None:
434
config.db = _existing_engine
435
_existing_engine = None
437
def startContext(self, ctx):
438
if not isinstance(ctx, type) \
439
or not issubclass(ctx, fixtures.TestBase):
442
self._setup_engine(ctx)
444
def stopContext(self, ctx):
445
if not isinstance(ctx, type) \
446
or not issubclass(ctx, fixtures.TestBase):
448
engines.testing_reaper._stop_test_ctx()
449
if not options.low_connections:
450
assertions.global_cleanup_assertions()
451
self._restore_engine(ctx)
added
removed
lib/sqlalchemy/testing/plugin/noseplugin.py
