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
|
#
# -*- coding: utf-8 -*-
# Extends unittest with support for pytest-style fixtures.
#
# Copyright (c) 2018 Peter Wu <peter@lekensteyn.nl>
#
# SPDX-License-Identifier: (GPL-2.0-or-later OR MIT)
#
import argparse
import functools
import inspect
import sys
import unittest
_use_native_pytest = False
def enable_pytest():
global _use_native_pytest, pytest
assert not _fallback
import pytest
_use_native_pytest = True
def fixture(scope="function", params=None, autouse=False, ids=None, name=None):
"""
When running under pytest, this is the same as the pytest.fixture decorator.
See https://docs.pytest.org/en/latest/reference.html#pytest-fixture
"""
if _use_native_pytest:
# XXX sorting of fixtures based on scope does not work, see
# https://github.com/pytest-dev/pytest/issues/4143#issuecomment-431794076
# When ran under pytest, use native functionality.
return pytest.fixture(scope, params, autouse, ids, name)
init_fallback_fixtures_once()
return _fallback.fixture(scope, params, autouse, ids, name)
def _fixture_wrapper(test_fn, params):
@functools.wraps(test_fn)
def wrapped(self):
if not _use_native_pytest:
self._fixture_request.function = getattr(self, test_fn.__name__)
self._fixture_request.fillfixtures(params)
fixtures = [self._fixture_request.getfixturevalue(n) for n in params]
test_fn(self, *fixtures)
return wrapped
def uses_fixtures(cls):
"""Enables use of fixtures within test methods of unittest.TestCase."""
assert issubclass(cls, unittest.TestCase)
for name in dir(cls):
func = getattr(cls, name)
if not name.startswith('test') or not callable(func):
continue
params = inspect.getfullargspec(func).args[1:]
# Unconditionally overwrite methods in case usefixtures marks exist.
setattr(cls, name, _fixture_wrapper(func, params))
if _use_native_pytest:
# Make request object to _fixture_wrapper
@pytest.fixture(autouse=True)
def __inject_request(self, request):
self._fixture_request = request
cls.__inject_request = __inject_request
else:
_patch_unittest_testcase_class(cls)
return cls
def mark_usefixtures(*args):
"""Add the given fixtures to every test method."""
if _use_native_pytest:
return pytest.mark.usefixtures(*args)
def wrapper(cls):
cls._fixtures_prepend = list(args)
return cls
return wrapper
# Begin fallback functionality when pytest is not available.
# Supported:
# - session-scoped fixtures (for cmd_tshark)
# - function-scoped fixtures (for tmpfile)
# - teardown (via yield keyword in fixture)
# - sorting of scopes (session before function)
# - fixtures that depend on other fixtures (requires sorting)
# - marking classes with @pytest.mark.usefixtures("fixture")
# Not supported (yet) due to lack of need for it:
# - autouse fixtures
# - parameterized fixtures (@pytest.fixture(params=...))
# - class-scoped fixtures
# - (overriding) fixtures on various levels (e.g. conftest, module, class)
class _FixtureSpec(object):
def __init__(self, name, scope, func):
self.name = name
self.scope = scope
self.func = func
self.params = inspect.getfullargspec(func).args
if inspect.ismethod(self.params):
self.params = self.params[1:] # skip self
def __repr__(self):
return '<_FixtureSpec name=%s scope=%s params=%r>' % \
(self.name, self.scope, self.params)
class _FixturesManager(object):
'''Records collected fixtures when pytest is unavailable.'''
fixtures = {}
# supported scopes, in execution order.
SCOPES = ('session', 'function')
def _add_fixture(self, scope, autouse, name, func):
name = name or func.__name__
if name in self.fixtures:
raise NotImplementedError('overriding fixtures is not supported')
self.fixtures[name] = _FixtureSpec(name, scope, func)
return func
def fixture(self, scope, params, autouse, ids, name):
if params:
raise NotImplementedError('params is not supported')
if ids:
raise NotImplementedError('ids is not supported')
if autouse:
raise NotImplementedError('autouse is not supported yet')
if callable(scope):
# used as decorator, pass through the original function
self._add_fixture('function', autouse, name, scope)
return scope
assert scope in self.SCOPES, 'unsupported scope'
# invoked with arguments, should return a decorator
return lambda func: self._add_fixture(scope, autouse, name, func)
def lookup(self, name):
return self.fixtures.get(name)
def resolve_fixtures(self, fixtures):
'''Find all dependencies for the requested list of fixtures.'''
unresolved = fixtures.copy()
resolved_keys, resolved = [], []
while unresolved:
param = unresolved.pop(0)
if param in resolved:
continue
spec = self.lookup(param)
if not spec:
if param == 'request':
continue
raise RuntimeError("Fixture '%s' not found" % (param,))
unresolved += spec.params
resolved_keys.append(param)
resolved.append(spec)
# Return fixtures, sorted by their scope
resolved.sort(key=lambda spec: self.SCOPES.index(spec.scope))
return resolved
class _ExecutionScope(object):
'''Store execution/teardown state for a scope.'''
def __init__(self, scope, parent):
self.scope = scope
self.parent = parent
self.cache = {}
self.finalizers = []
def _find_scope(self, scope):
context = self
while context.scope != scope:
context = context.parent
return context
def execute(self, spec, test_fn):
'''Execute a fixture and cache the result.'''
context = self._find_scope(spec.scope)
if spec.name in context.cache:
return
try:
value, cleanup = self._execute_one(spec, test_fn)
exc = None
except Exception:
value, cleanup, exc = None, None, sys.exc_info()[1]
context.cache[spec.name] = value, exc
if cleanup:
context.finalizers.append(cleanup)
if exc:
raise exc
def cached_result(self, spec):
'''Obtain the cached result for a previously executed fixture.'''
entry = self._find_scope(spec.scope).cache.get(spec.name)
if not entry:
return None, False
value, exc = entry
if exc:
raise exc
return value, True
def _execute_one(self, spec, test_fn):
# A fixture can only execute in the same or earlier scopes
context_scope_index = _FixturesManager.SCOPES.index(self.scope)
fixture_scope_index = _FixturesManager.SCOPES.index(spec.scope)
assert fixture_scope_index <= context_scope_index
if spec.params:
# Do not invoke destroy, it is taken care of by the main request.
subrequest = _FixtureRequest(self)
subrequest.function = test_fn
subrequest.fillfixtures(spec.params)
fixtures = (subrequest.getfixturevalue(n) for n in spec.params)
value = spec.func(*fixtures) # Execute fixture
else:
value = spec.func() # Execute fixture
if not inspect.isgenerator(value):
return value, None
@functools.wraps(value)
def cleanup():
try:
next(value)
except StopIteration:
pass
else:
raise RuntimeError('%s yielded more than once!' % (spec.name,))
return next(value), cleanup
def destroy(self):
exceptions = []
for cleanup in self.finalizers:
try:
cleanup()
except:
exceptions.append(sys.exc_info()[1])
self.cache.clear()
self.finalizers.clear()
if exceptions:
raise exceptions[0]
class _FixtureRequest(object):
'''
Holds state during a single test execution. See
https://docs.pytest.org/en/latest/reference.html#request
'''
def __init__(self, context):
self._context = context
self._fixtures_prepend = [] # fixtures added via usefixtures
# XXX is there any need for .module or .cls?
self.function = None # test function, set before execution.
def fillfixtures(self, params):
params = self._fixtures_prepend + params
specs = _fallback.resolve_fixtures(params)
for spec in specs:
self._context.execute(spec, self.function)
def getfixturevalue(self, argname):
spec = _fallback.lookup(argname)
if not spec:
assert argname == 'request'
return self
value, ok = self._context.cached_result(spec)
if not ok:
# If getfixturevalue is called directly from a setUp function, the
# fixture value might not have computed before, so evaluate it now.
# As the test function is not available, use None.
self._context.execute(spec, test_fn=None)
value, ok = self._context.cached_result(spec)
assert ok, 'Failed to execute fixture %s' % (spec,)
return value
def destroy(self):
self._context.destroy()
def addfinalizer(self, finalizer):
self._context.finalizers.append(finalizer)
@property
def instance(self):
return self.function.__self__
@property
def config(self):
'''The pytest config object associated with this request.'''
return _config
def _patch_unittest_testcase_class(cls):
'''
Patch the setUp and tearDown methods of the unittest.TestCase such that the
fixtures are properly setup and destroyed.
'''
def setUp(self):
assert _session_context, 'must call create_session() first!'
function_context = _ExecutionScope('function', _session_context)
req = _FixtureRequest(function_context)
req._fixtures_prepend = getattr(self, '_fixtures_prepend', [])
self._fixture_request = req
self._orig_setUp()
def tearDown(self):
try:
self._orig_tearDown()
finally:
self._fixture_request.destroy()
# Only the leaf test case class should be decorated!
assert not hasattr(cls, '_orig_setUp')
assert not hasattr(cls, '_orig_tearDown')
cls._orig_setUp, cls.setUp = cls.setUp, setUp
cls._orig_tearDown, cls.tearDown = cls.tearDown, tearDown
class _Config(object):
def __init__(self, args):
assert isinstance(args, argparse.Namespace)
self.args = args
def getoption(self, name, default):
'''Partial emulation for pytest Config.getoption.'''
name = name.lstrip('-').replace('-', '_')
return getattr(self.args, name, default)
_fallback = None
_session_context = None
_config = None
def init_fallback_fixtures_once():
global _fallback
assert not _use_native_pytest
if _fallback:
return
_fallback = _FixturesManager()
# Register standard fixtures here as needed
def create_session(args=None):
'''Start a test session where args is from argparse.'''
global _session_context, _config
assert not _use_native_pytest
_session_context = _ExecutionScope('session', None)
if args is None:
args = argparse.Namespace()
_config = _Config(args)
def destroy_session():
global _session_context
assert not _use_native_pytest
_session_context = None
def skip(msg):
'''Skip the executing test with the given message.'''
if _use_native_pytest:
pytest.skip(msg)
else:
raise unittest.SkipTest(msg)
|