← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to lib/sqlalchemy/testing/exclusions.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/sqlalchemy/testing/exclusions.py
 
1
from __future__ import with_statement
 
2
 
 
3
import operator
 
4
from nose import SkipTest
 
5
from ..util import decorator
 
6
from . import config
 
7
from .. import util
 
8
import contextlib
 
9
 
 
10
 
 
11
class skip_if(object):
 
12
    def __init__(self, predicate, reason=None):
 
13
        self.predicate = _as_predicate(predicate)
 
14
        self.reason = reason
 
15
 
 
16
    _fails_on = None
 
17
 
 
18
    @property
 
19
    def enabled(self):
 
20
        return not self.predicate()
 
21
 
 
22
    @contextlib.contextmanager
 
23
    def fail_if(self, name='block'):
 
24
        try:
 
25
            yield
 
26
        except Exception, ex:
 
27
            if self.predicate():
 
28
                print ("%s failed as expected (%s): %s " % (
 
29
                    name, self.predicate, str(ex)))
 
30
            else:
 
31
                raise
 
32
        else:
 
33
            if self.predicate():
 
34
                raise AssertionError(
 
35
                    "Unexpected success for '%s' (%s)" %
 
36
                    (name, self.predicate))
 
37
 
 
38
    def __call__(self, fn):
 
39
        @decorator
 
40
        def decorate(fn, *args, **kw):
 
41
            if self.predicate():
 
42
                if self.reason:
 
43
                    msg = "'%s' : %s" % (
 
44
                            fn.__name__,
 
45
                            self.reason
 
46
                        )
 
47
                else:
 
48
                    msg = "'%s': %s" % (
 
49
                            fn.__name__, self.predicate
 
50
                        )
 
51
                raise SkipTest(msg)
 
52
            else:
 
53
                if self._fails_on:
 
54
                    with self._fails_on.fail_if(name=fn.__name__):
 
55
                        return fn(*args, **kw)
 
56
                else:
 
57
                    return fn(*args, **kw)
 
58
        return decorate(fn)
 
59
 
 
60
    def fails_on(self, other, reason=None):
 
61
        self._fails_on = skip_if(other, reason)
 
62
        return self
 
63
 
 
64
 
 
65
class fails_if(skip_if):
 
66
    def __call__(self, fn):
 
67
        @decorator
 
68
        def decorate(fn, *args, **kw):
 
69
            with self.fail_if(name=fn.__name__):
 
70
                return fn(*args, **kw)
 
71
        return decorate(fn)
 
72
 
 
73
 
 
74
def only_if(predicate, reason=None):
 
75
    predicate = _as_predicate(predicate)
 
76
    return skip_if(NotPredicate(predicate), reason)
 
77
 
 
78
 
 
79
def succeeds_if(predicate, reason=None):
 
80
    predicate = _as_predicate(predicate)
 
81
    return fails_if(NotPredicate(predicate), reason)
 
82
 
 
83
 
 
84
class Predicate(object):
 
85
    @classmethod
 
86
    def as_predicate(cls, predicate):
 
87
        if isinstance(predicate, skip_if):
 
88
            return predicate.predicate
 
89
        elif isinstance(predicate, Predicate):
 
90
            return predicate
 
91
        elif isinstance(predicate, list):
 
92
            return OrPredicate([cls.as_predicate(pred) for pred in predicate])
 
93
        elif isinstance(predicate, tuple):
 
94
            return SpecPredicate(*predicate)
 
95
        elif isinstance(predicate, basestring):
 
96
            return SpecPredicate(predicate, None, None)
 
97
        elif util.callable(predicate):
 
98
            return LambdaPredicate(predicate)
 
99
        else:
 
100
            assert False, "unknown predicate type: %s" % predicate
 
101
 
 
102
 
 
103
class BooleanPredicate(Predicate):
 
104
    def __init__(self, value, description=None):
 
105
        self.value = value
 
106
        self.description = description or "boolean %s" % value
 
107
 
 
108
    def __call__(self):
 
109
        return self.value
 
110
 
 
111
    def _as_string(self, negate=False):
 
112
        if negate:
 
113
            return "not " + self.description
 
114
        else:
 
115
            return self.description
 
116
 
 
117
    def __str__(self):
 
118
        return self._as_string()
 
119
 
 
120
 
 
121
class SpecPredicate(Predicate):
 
122
    def __init__(self, db, op=None, spec=None, description=None):
 
123
        self.db = db
 
124
        self.op = op
 
125
        self.spec = spec
 
126
        self.description = description
 
127
 
 
128
    _ops = {
 
129
            '<': operator.lt,
 
130
             '>': operator.gt,
 
131
             '==': operator.eq,
 
132
             '!=': operator.ne,
 
133
             '<=': operator.le,
 
134
             '>=': operator.ge,
 
135
             'in': operator.contains,
 
136
             'between': lambda val, pair: val >= pair[0] and val <= pair[1],
 
137
             }
 
138
 
 
139
    def __call__(self, engine=None):
 
140
        if engine is None:
 
141
            engine = config.db
 
142
 
 
143
        if "+" in self.db:
 
144
            dialect, driver = self.db.split('+')
 
145
        else:
 
146
            dialect, driver = self.db, None
 
147
 
 
148
        if dialect and engine.name != dialect:
 
149
            return False
 
150
        if driver is not None and engine.driver != driver:
 
151
            return False
 
152
 
 
153
        if self.op is not None:
 
154
            assert driver is None, "DBAPI version specs not supported yet"
 
155
 
 
156
            version = _server_version(engine)
 
157
            oper = hasattr(self.op, '__call__') and self.op \
 
158
                        or self._ops[self.op]
 
159
            return oper(version, self.spec)
 
160
        else:
 
161
            return True
 
162
 
 
163
    def _as_string(self, negate=False):
 
164
        if self.description is not None:
 
165
            return self.description
 
166
        elif self.op is None:
 
167
            if negate:
 
168
                return "not %s" % self.db
 
169
            else:
 
170
                return "%s" % self.db
 
171
        else:
 
172
            if negate:
 
173
                return "not %s %s %s" % (
 
174
                        self.db,
 
175
                        self.op,
 
176
                        self.spec
 
177
                    )
 
178
            else:
 
179
                return "%s %s %s" % (
 
180
                        self.db,
 
181
                        self.op,
 
182
                        self.spec
 
183
                    )
 
184
 
 
185
    def __str__(self):
 
186
        return self._as_string()
 
187
 
 
188
 
 
189
class LambdaPredicate(Predicate):
 
190
    def __init__(self, lambda_, description=None, args=None, kw=None):
 
191
        self.lambda_ = lambda_
 
192
        self.args = args or ()
 
193
        self.kw = kw or {}
 
194
        if description:
 
195
            self.description = description
 
196
        elif lambda_.__doc__:
 
197
            self.description = lambda_.__doc__
 
198
        else:
 
199
            self.description = "custom function"
 
200
 
 
201
    def __call__(self):
 
202
        return self.lambda_(*self.args, **self.kw)
 
203
 
 
204
    def _as_string(self, negate=False):
 
205
        if negate:
 
206
            return "not " + self.description
 
207
        else:
 
208
            return self.description
 
209
 
 
210
    def __str__(self):
 
211
        return self._as_string()
 
212
 
 
213
 
 
214
class NotPredicate(Predicate):
 
215
    def __init__(self, predicate):
 
216
        self.predicate = predicate
 
217
 
 
218
    def __call__(self, *arg, **kw):
 
219
        return not self.predicate(*arg, **kw)
 
220
 
 
221
    def __str__(self):
 
222
        return self.predicate._as_string(True)
 
223
 
 
224
 
 
225
class OrPredicate(Predicate):
 
226
    def __init__(self, predicates, description=None):
 
227
        self.predicates = predicates
 
228
        self.description = description
 
229
 
 
230
    def __call__(self, *arg, **kw):
 
231
        for pred in self.predicates:
 
232
            if pred(*arg, **kw):
 
233
                self._str = pred
 
234
                return True
 
235
        return False
 
236
 
 
237
    _str = None
 
238
 
 
239
    def _eval_str(self, negate=False):
 
240
        if self._str is None:
 
241
            if negate:
 
242
                conjunction = " and "
 
243
            else:
 
244
                conjunction = " or "
 
245
            return conjunction.join(p._as_string(negate=negate)
 
246
                            for p in self.predicates)
 
247
        else:
 
248
            return self._str._as_string(negate=negate)
 
249
 
 
250
    def _negation_str(self):
 
251
        if self.description is not None:
 
252
            return "Not " + (self.description % {"spec": self._str})
 
253
        else:
 
254
            return self._eval_str(negate=True)
 
255
 
 
256
    def _as_string(self, negate=False):
 
257
        if negate:
 
258
            return self._negation_str()
 
259
        else:
 
260
            if self.description is not None:
 
261
                return self.description % {"spec": self._str}
 
262
            else:
 
263
                return self._eval_str()
 
264
 
 
265
    def __str__(self):
 
266
        return self._as_string()
 
267
 
 
268
_as_predicate = Predicate.as_predicate
 
269
 
 
270
 
 
271
def _is_excluded(db, op, spec):
 
272
    return SpecPredicate(db, op, spec)()
 
273
 
 
274
 
 
275
def _server_version(engine):
 
276
    """Return a server_version_info tuple."""
 
277
 
 
278
    # force metadata to be retrieved
 
279
    conn = engine.connect()
 
280
    version = getattr(engine.dialect, 'server_version_info', ())
 
281
    conn.close()
 
282
    return version
 
283
 
 
284
 
 
285
def db_spec(*dbs):
 
286
    return OrPredicate(
 
287
            Predicate.as_predicate(db) for db in dbs
 
288
        )
 
289
 
 
290
 
 
291
def open():
 
292
    return skip_if(BooleanPredicate(False, "mark as execute"))
 
293
 
 
294
 
 
295
def closed():
 
296
    return skip_if(BooleanPredicate(True, "marked as skip"))
 
297
 
 
298
 
 
299
@decorator
 
300
def future(fn, *args, **kw):
 
301
    return fails_if(LambdaPredicate(fn, *args, **kw), "Future feature")
 
302
 
 
303
 
 
304
def fails_on(db, reason=None):
 
305
    return fails_if(SpecPredicate(db), reason)
 
306
 
 
307
 
 
308
def fails_on_everything_except(*dbs):
 
309
    return succeeds_if(
 
310
                OrPredicate([
 
311
                    SpecPredicate(db) for db in dbs
 
312
                    ])
 
313
            )
 
314
 
 
315
 
 
316
def skip(db, reason=None):
 
317
    return skip_if(SpecPredicate(db), reason)
 
318
 
 
319
 
 
320
def only_on(dbs, reason=None):
 
321
    return only_if(
 
322
            OrPredicate([SpecPredicate(db) for db in util.to_list(dbs)])
 
323
    )
 
324
 
 
325
 
 
326
def exclude(db, op, spec, reason=None):
 
327
    return skip_if(SpecPredicate(db, op, spec), reason)
 
328
 
 
329
 
 
330
def against(*queries):
 
331
    return OrPredicate([
 
332
                Predicate.as_predicate(query)
 
333
                for query in queries
 
334
            ])()