2
from nose import SkipTest
3
from sqlalchemy.util import decorator
4
from test.bootstrap import config
5
from sqlalchemy import util
8
def fails_if(predicate, reason=None):
9
predicate = _as_predicate(predicate)
12
def decorate(fn, *args, **kw):
14
return fn(*args, **kw)
19
print ("'%s' failed as expected (%s): %s " % (
20
fn.__name__, predicate, str(ex)))
24
"Unexpected success for '%s' (%s)" %
25
(fn.__name__, predicate))
28
def skip_if(predicate, reason=None):
29
predicate = _as_predicate(predicate)
32
def decorate(fn, *args, **kw):
41
fn.__name__, predicate
45
return fn(*args, **kw)
48
def only_if(predicate, reason=None):
49
predicate = _as_predicate(predicate)
50
return skip_if(NotPredicate(predicate), reason)
52
def succeeds_if(predicate, reason=None):
53
predicate = _as_predicate(predicate)
54
return fails_if(NotPredicate(predicate), reason)
56
class Predicate(object):
58
def as_predicate(cls, predicate):
59
if isinstance(predicate, Predicate):
61
elif isinstance(predicate, list):
62
return OrPredicate([cls.as_predicate(pred) for pred in predicate])
63
elif isinstance(predicate, tuple):
64
return SpecPredicate(*predicate)
65
elif isinstance(predicate, basestring):
66
return SpecPredicate(predicate, None, None)
67
elif util.callable(predicate):
68
return LambdaPredicate(predicate)
70
assert False, "unknown predicate type: %s" % predicate
72
class SpecPredicate(Predicate):
73
def __init__(self, db, op=None, spec=None, description=None):
77
self.description = description
86
'in': operator.contains,
87
'between': lambda val, pair: val >= pair[0] and val <= pair[1],
90
def __call__(self, engine=None):
95
dialect, driver = self.db.split('+')
97
dialect, driver = self.db, None
99
if dialect and engine.name != dialect:
101
if driver is not None and engine.driver != driver:
104
if self.op is not None:
105
assert driver is None, "DBAPI version specs not supported yet"
107
version = _server_version()
108
oper = hasattr(self.op, '__call__') and self.op \
109
or self._ops[self.op]
110
return oper(version, self.spec)
114
def _as_string(self, negate=False):
115
if self.description is not None:
116
return self.description
117
elif self.op is None:
119
return "not %s" % self.db
121
return "%s" % self.db
124
return "not %s %s %s" % (
130
return "%s %s %s" % (
137
return self._as_string()
139
class LambdaPredicate(Predicate):
140
def __init__(self, lambda_, description=None, args=None, kw=None):
141
self.lambda_ = lambda_
142
self.args = args or ()
145
self.description = description
146
elif lambda_.__doc__:
147
self.description = lambda_.__doc__
149
self.description = "custom function"
152
return self.lambda_(*self.args, **self.kw)
154
def _as_string(self, negate=False):
156
return "not " + self.description
158
return self.description
161
return self._as_string()
163
class NotPredicate(Predicate):
164
def __init__(self, predicate):
165
self.predicate = predicate
167
def __call__(self, *arg, **kw):
168
return not self.predicate(*arg, **kw)
171
return self.predicate._as_string(True)
173
class OrPredicate(Predicate):
174
def __init__(self, predicates, description=None):
175
self.predicates = predicates
176
self.description = description
178
def __call__(self, *arg, **kw):
179
for pred in self.predicates:
187
def _eval_str(self, negate=False):
188
if self._str is None:
190
conjunction = " and "
193
return conjunction.join(p._as_string(negate=negate)
194
for p in self.predicates)
196
return self._str._as_string(negate=negate)
198
def _negation_str(self):
199
if self.description is not None:
200
return "Not " + (self.description % {"spec": self._str})
202
return self._eval_str(negate=True)
204
def _as_string(self, negate=False):
206
return self._negation_str()
208
if self.description is not None:
209
return self.description % {"spec": self._str}
211
return self._eval_str()
214
return self._as_string()
216
_as_predicate = Predicate.as_predicate
218
def _is_excluded(db, op, spec):
219
return SpecPredicate(db, op, spec)()
221
def _server_version(bind=None):
222
"""Return a server_version_info tuple."""
227
# force metadata to be retrieved
228
conn = bind.connect()
229
version = getattr(bind.dialect, 'server_version_info', ())
235
Predicate.as_predicate(db) for db in dbs
239
def future(fn, *args, **kw):
240
return fails_if(LambdaPredicate(fn, *args, **kw), "Future feature")
242
def fails_on(db, reason):
243
return fails_if(SpecPredicate(db), reason)
245
def fails_on_everything_except(*dbs):
248
SpecPredicate(db) for db in dbs
252
def skip(db, reason):
253
return skip_if(SpecPredicate(db), reason)
255
def only_on(dbs, reason):
257
OrPredicate([SpecPredicate(db) for db in util.to_list(dbs)])
261
def exclude(db, op, spec, reason):
262
return skip_if(SpecPredicate(db, op, spec), reason)
265
def against(*queries):
267
Predicate.as_predicate(query)
added
removed
test/lib/exclusions.py
