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
|
"""Enhance nose with extra options and behaviors for running SQLAlchemy tests.
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 library, 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 <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="sqlite",
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', os.path.expanduser('~/.satest.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('test.lib') or \
fn.__module__.startswith('test.bootstrap'):
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)
check(test_suite)()
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()
|
