Source

sqlalchemy / lib / sqlalchemy / testing / plugin / plugin_base.py

Full commit
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
# plugin/plugin_base.py
# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php

"""Testing extensions.

this module is designed to work as a testing-framework-agnostic library,
so that we can continue to support nose and also begin adding new functionality
via py.test.

"""

from __future__ import absolute_import
try:
    # needs 2.7...
    from unittest import SkipTest
except ImportError:
    try:
        from nose import SkipTest
    except ImportError:
        from _pytest.runner import Skipped as SkipTest

import sys
import re

py3k = sys.version_info >= (3, 0)

if py3k:
    import configparser
else:
    import ConfigParser as configparser


# late imports
fixtures = None
engines = None
exclusions = None
warnings = None
profiling = None
assertions = None
requirements = None
config = None
testing = None
util = None
file_config = None


logging = None
db_opts = {}
options = None

def setup_options(make_option):
    make_option("--log-info", action="callback", type="string", callback=_log,
        help="turn on info logging for <LOG> (multiple OK)")
    make_option("--log-debug", action="callback", type="string", callback=_log,
        help="turn on debug logging for <LOG> (multiple OK)")
    make_option("--db", action="append", type="string", dest="db",
                help="Use prefab database uri. Multiple OK, "
                        "first one is run by default.")
    make_option('--dbs', action='callback', callback=_list_dbs,
        help="List available prefab dbs")
    make_option("--dburi", action="append", type="string", dest="dburi",
        help="Database uri.  Multiple OK, first one is run by default.")
    make_option("--dropfirst", action="store_true", dest="dropfirst",
        help="Drop all tables in the target database first")
    make_option("--mockpool", action="store_true", dest="mockpool",
        help="Use mock pool (asserts only one connection used)")
    make_option("--low-connections", action="store_true", dest="low_connections",
        help="Use a low number of distinct connections - i.e. for Oracle TNS"
    )
    make_option("--reversetop", action="store_true", dest="reversetop", default=False,
        help="Use a random-ordering set implementation in the ORM (helps "
              "reveal dependency issues)")
    make_option("--requirements", action="callback", type="string",
        callback=_requirements_opt,
        help="requirements class for testing, overrides setup.cfg")
    make_option("--with-cdecimal", action="store_true", dest="cdecimal", default=False,
        help="Monkeypatch the cdecimal library into Python 'decimal' for all tests")
    make_option("--serverside", action="callback", callback=_server_side_cursors,
        help="Turn on server side cursors for PG")
    make_option("--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.")
    make_option("--tableopts", action="append", dest="tableopts", default=[],
        help="Add a dialect-specific table option, key=value")
    make_option("--write-profiles", action="store_true", dest="write_profiles", default=False,
            help="Write/update profiling data.")

def read_config():
    global file_config
    file_config = configparser.ConfigParser()
    file_config.read(['setup.cfg', 'test.cfg'])

def pre_begin(opt):
    """things to set up early, before coverage might be setup."""
    global options
    options = opt
    for fn in pre_configure:
        fn(options, file_config)

def set_coverage_flag(value):
    options.has_coverage = value

def post_begin():
    """things to set up later, once we know coverage is running."""
    # Lazy setup of other options (post coverage)
    for fn in post_configure:
        fn(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,\
                    config, testing
    from sqlalchemy import testing
    from sqlalchemy.testing import fixtures, engines, exclusions, \
                    assertions, warnings, profiling, config
    from sqlalchemy import util


def _log(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(opt_str, value, parser):
    db_opts['server_side_cursors'] = True

def _requirements_opt(opt_str, value, parser):
    _setup_requirements(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):
    from sqlalchemy.testing import engines, config
    from sqlalchemy import testing

    if options.dburi:
        db_urls = list(options.dburi)
    else:
        db_urls = []

    if options.db:
        for db_token in options.db:
            for db in re.split(r'[,\s]+', db_token):
                if db not in file_config.options('db'):
                    raise RuntimeError(
                        "Unknown URI specifier '%s'.  Specify --dbs for known uris."
                                % db)
                else:
                    db_urls.append(file_config.get('db', db))

    if not db_urls:
        db_urls.append(file_config.get('db', 'default'))

    for db_url in db_urls:
        eng = engines.testing_engine(db_url, db_opts)
        eng.connect().close()
        config.Config.register(eng, db_opts, options, file_config, testing)

    config.db_opts = db_opts


@post
def _engine_pool(options, file_config):
    if options.mockpool:
        from sqlalchemy import pool
        db_opts['poolclass'] = pool.AssertionPool

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

@post
def _prep_testing_database(options, file_config):
    from sqlalchemy.testing import config
    from sqlalchemy import schema, inspect

    if options.dropfirst:
        for cfg in config.Config.all_configs():
            e = cfg.db
            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())))

            if config.requirements.schemas.enabled_for_config(cfg):
                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())))

            if config.requirements.schemas.enabled_for_config(cfg):
                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")))


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


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


def want_class(cls):
    if not issubclass(cls, fixtures.TestBase):
        return False
    elif cls.__name__.startswith('_'):
        return False
    else:
        return True

def generate_sub_tests(cls, module):
    if getattr(cls, '__multiple__', False):
        for cfg in config.Config.all_configs():
            name = "%s_%s_%s" % (cls.__name__, cfg.db.name, cfg.db.driver)
            subcls = type(
                        name,
                        (cls, ),
                        {
                            "__only_on__": (cfg.db.name, cfg.db.driver),
                            "__multiple__": False}
                        )
            setattr(module, name, subcls)
            yield subcls
    else:
        yield cls


def start_test_class(cls):
    _do_skips(cls)
    _setup_engine(cls)

def stop_test_class(cls):
    engines.testing_reaper._stop_test_ctx()
    if not options.low_connections:
        assertions.global_cleanup_assertions()
    _restore_engine()

def _restore_engine():
    config._current.reset(testing)

def _setup_engine(cls):
    if getattr(cls, '__engine_options__', None):
        eng = engines.testing_engine(options=cls.__engine_options__)
        config._current.push_engine(eng, testing)

def before_test(test, id_):
    warnings.resetwarnings()
    profiling._current_test = id_

def after_test(test):
    engines.testing_reaper._after_test_ctx()
    warnings.resetwarnings()

def _do_skips(cls):
    all_configs = set(config.Config.all_configs())
    reasons = []

    if hasattr(cls, '__requires__'):
        requirements = config.requirements
        for config_obj in list(all_configs):
            for requirement in cls.__requires__:
                check = getattr(requirements, requirement)

                if check.predicate(config_obj):
                    all_configs.remove(config_obj)
                    if check.reason:
                        reasons.append(check.reason)
                    break

    if cls.__unsupported_on__:
        spec = exclusions.db_spec(*cls.__unsupported_on__)
        for config_obj in list(all_configs):
            if spec(config_obj):
                all_configs.remove(config_obj)

    if getattr(cls, '__only_on__', None):
        spec = exclusions.db_spec(*util.to_list(cls.__only_on__))
        for config_obj in list(all_configs):
            if not spec(config_obj):
                all_configs.remove(config_obj)


    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_spec, op, spec in getattr(cls, '__excluded_on__', ()):
        for config_obj in list(all_configs):
            if exclusions.skip_if(
                    exclusions.SpecPredicate(db_spec, op, spec)
                    ).predicate(config_obj):
                all_configs.remove(config_obj)


    if not all_configs:
        raise SkipTest(
            "'%s' unsupported on DB implementation %s%s" % (
                cls.__name__,
                ", ".join("'%s' = %s" % (
                                config_obj.db.name,
                                config_obj.db.dialect.server_version_info)
                    for config_obj in config.Config.all_configs()
                ),
                ", ".join(reasons)
            )
        )
    elif hasattr(cls, '__prefer__'):
        non_preferred = set()
        spec = exclusions.db_spec(*util.to_list(cls.__prefer__))
        for config_obj in all_configs:
            if not spec(config_obj):
                non_preferred.add(config_obj)
        if all_configs.difference(non_preferred):
            all_configs.difference_update(non_preferred)

    if config._current not in all_configs:
        _setup_config(all_configs.pop(), cls)

def _setup_config(config_obj, ctx):
    config._current.push(config_obj, testing)