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
|
import logging
import os
import re
import sys
import time
import warnings
import ConfigParser
import StringIO
from config import db, db_label, db_url, file_config, base_config, \
post_configure, \
_list_dbs, _server_side_cursors, _engine_strategy, \
_engine_uri, _require, _engine_pool, \
_create_testing_engine, _prep_testing_database, \
_set_table_options, _reverse_topological, _log
from sqlalchemy.test import testing, config, requires
from nose.plugins import Plugin
from nose.util import tolist
import nose.case
log = logging.getLogger('nose.plugins.sqlalchemy')
class NoseSQLAlchemy(Plugin):
"""
Handles the setup and extra properties required for testing SQLAlchemy
"""
enabled = True
name = 'sqlalchemy'
score = 100
def options(self, parser, env=os.environ):
Plugin.options(self, parser, env)
opt = parser.add_option
#opt("--verbose", action="store_true", dest="verbose",
#help="enable stdout echoing/printing")
#opt("--quiet", action="store_true", dest="quiet", help="suppress output")
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 (use with caution on Oracle, MS-SQL)")
opt("--mockpool", action="store_true", dest="mockpool",
help="Use mock pool (asserts only one connection used)")
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="Reverse the collection ordering for topological sorts (helps "
"reveal dependency issues)")
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")
global file_config
file_config = ConfigParser.ConfigParser()
file_config.readfp(StringIO.StringIO(base_config))
file_config.read(['test.cfg', os.path.expanduser('~/.satest.cfg')])
config.file_config = file_config
def configure(self, options, conf):
Plugin.configure(self, options, conf)
import testing, requires
testing.db = db
testing.requires = requires
# Lazy setup of other options (post coverage)
for fn in post_configure:
fn(options, file_config)
def describeTest(self, test):
return ""
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, testing.TestBase):
return False
else:
if (hasattr(cls, '__whitelist__') and
testing.db.name in cls.__whitelist__):
return True
else:
return not self.__should_skip_for(cls)
def __should_skip_for(self, cls):
if hasattr(cls, '__requires__'):
def test_suite(): return 'ok'
for requirement in cls.__requires__:
check = getattr(requires, requirement)
if check(test_suite)() != 'ok':
# The requirement will perform messaging.
return True
if (hasattr(cls, '__unsupported_on__') and
testing.db.name in cls.__unsupported_on__):
print "'%s' unsupported on DB implementation '%s'" % (
cls.__class__.__name__, testing.db.name)
return True
if (getattr(cls, '__only_on__', None) not in (None, testing.db.name)):
print "'%s' unsupported on DB implementation '%s'" % (
cls.__class__.__name__, testing.db.name)
return True
if (getattr(cls, '__skip_if__', False)):
for c in getattr(cls, '__skip_if__'):
if c():
print "'%s' skipped by %s" % (
cls.__class__.__name__, c.__name__)
return True
for rule in getattr(cls, '__excluded_on__', ()):
if testing._is_excluded(*rule):
print "'%s' unsupported on DB %s version %s" % (
cls.__class__.__name__, testing.db.name,
_server_version())
return True
return False
#def begin(self):
#pass
def beforeTest(self, test):
testing.resetwarnings()
def afterTest(self, test):
testing.resetwarnings()
#def handleError(self, test, err):
#pass
#def finalize(self, result=None):
#pass
|
