test/fixtures.py

   1 #
   2 # -*- coding: utf-8 -*-
   3 # Extends unittest with support for pytest-style fixtures.
   4 #
   5 # Copyright (c) 2018 Peter Wu <peter@lekensteyn.nl>
   6 #
   7 # SPDX-License-Identifier: (GPL-2.0-or-later OR MIT)
   8 #
   9
  10 import argparse
  11 import functools
  12 import inspect
  13 import sys
  14 import unittest
  15
  16 _use_native_pytest = False
  17
  18
  19 def enable_pytest():
  20     global _use_native_pytest, pytest
  21     assert not _fallback
  22     import pytest
  23     _use_native_pytest = True
  24
  25
  26 def fixture(scope="function", params=None, autouse=False, ids=None, name=None):
  27     """
  28     When running under pytest, this is the same as the pytest.fixture decorator.
  29     See https://docs.pytest.org/en/latest/reference.html#pytest-fixture
  30     """
  31     if _use_native_pytest:
  32         # XXX sorting of fixtures based on scope does not work, see
  33         # https://github.com/pytest-dev/pytest/issues/4143#issuecomment-431794076
  34         # When ran under pytest, use native functionality.
  35         return pytest.fixture(scope, params, autouse, ids, name)
  36     init_fallback_fixtures_once()
  37     return _fallback.fixture(scope, params, autouse, ids, name)
  38
  39
  40 def _fixture_wrapper(test_fn, params):
  41     @functools.wraps(test_fn)
  42     def wrapped(self):
  43         if not _use_native_pytest:
  44             self._fixture_request.function = getattr(self, test_fn.__name__)
  45             self._fixture_request.fillfixtures(params)
  46         fixtures = [self._fixture_request.getfixturevalue(n) for n in params]
  47         test_fn(self, *fixtures)
  48     return wrapped
  49
  50
  51 def uses_fixtures(cls):
  52     """Enables use of fixtures within test methods of unittest.TestCase."""
  53     assert issubclass(cls, unittest.TestCase)
  54
  55     for name in dir(cls):
  56         func = getattr(cls, name)
  57         if not name.startswith('test') or not callable(func):
  58             continue
  59         params = inspect.getfullargspec(func).args[1:]
  60         # Unconditionally overwrite methods in case usefixtures marks exist.
  61         setattr(cls, name, _fixture_wrapper(func, params))
  62
  63     if _use_native_pytest:
  64         # Make request object to _fixture_wrapper
  65         @pytest.fixture(autouse=True)
  66         def __inject_request(self, request):
  67             self._fixture_request = request
  68         cls.__inject_request = __inject_request
  69     else:
  70         _patch_unittest_testcase_class(cls)
  71
  72     return cls
  73
  74
  75 def mark_usefixtures(*args):
  76     """Add the given fixtures to every test method."""
  77     if _use_native_pytest:
  78         return pytest.mark.usefixtures(*args)
  79
  80     def wrapper(cls):
  81         cls._fixtures_prepend = list(args)
  82         return cls
  83     return wrapper
  84
  85
  86 # Begin fallback functionality when pytest is not available.
  87 # Supported:
  88 # - session-scoped fixtures (for cmd_tshark)
  89 # - function-scoped fixtures (for tmpfile)
  90 # - teardown (via yield keyword in fixture)
  91 # - sorting of scopes (session before function)
  92 # - fixtures that depend on other fixtures (requires sorting)
  93 # - marking classes with @pytest.mark.usefixtures("fixture")
  94 # Not supported (yet) due to lack of need for it:
  95 # - autouse fixtures
  96 # - parameterized fixtures (@pytest.fixture(params=...))
  97 # - class-scoped fixtures
  98 # - (overriding) fixtures on various levels (e.g. conftest, module, class)
  99
 100
 101 class _FixtureSpec(object):
 102     def __init__(self, name, scope, func):
 103         self.name = name
 104         self.scope = scope
 105         self.func = func
 106         self.params = inspect.getfullargspec(func).args
 107         if inspect.ismethod(self.params):
 108             self.params = self.params[1:]  # skip self
 109
 110     def __repr__(self):
 111         return '<_FixtureSpec name=%s scope=%s params=%r>' % \
 112             (self.name, self.scope, self.params)
 113
 114
 115 class _FixturesManager(object):
 116     '''Records collected fixtures when pytest is unavailable.'''
 117     fixtures = {}
 118     # supported scopes, in execution order.
 119     SCOPES = ('session', 'function')
 120
 121     def _add_fixture(self, scope, autouse, name, func):
 122         name = name or func.__name__
 123         if name in self.fixtures:
 124             raise NotImplementedError('overriding fixtures is not supported')
 125         self.fixtures[name] = _FixtureSpec(name, scope, func)
 126         return func
 127
 128     def fixture(self, scope, params, autouse, ids, name):
 129         if params:
 130             raise NotImplementedError('params is not supported')
 131         if ids:
 132             raise NotImplementedError('ids is not supported')
 133         if autouse:
 134             raise NotImplementedError('autouse is not supported yet')
 135
 136         if callable(scope):
 137             # used as decorator, pass through the original function
 138             self._add_fixture('function', autouse, name, scope)
 139             return scope
 140         assert scope in self.SCOPES, 'unsupported scope'
 141         # invoked with arguments, should return a decorator
 142         return lambda func: self._add_fixture(scope, autouse, name, func)
 143
 144     def lookup(self, name):
 145         return self.fixtures.get(name)
 146
 147     def resolve_fixtures(self, fixtures):
 148         '''Find all dependencies for the requested list of fixtures.'''
 149         unresolved = fixtures.copy()
 150         resolved_keys, resolved = [], []
 151         while unresolved:
 152             param = unresolved.pop(0)
 153             if param in resolved:
 154                 continue
 155             spec = self.lookup(param)
 156             if not spec:
 157                 if param == 'request':
 158                     continue
 159                 raise RuntimeError("Fixture '%s' not found" % (param,))
 160             unresolved += spec.params
 161             resolved_keys.append(param)
 162             resolved.append(spec)
 163         # Return fixtures, sorted by their scope
 164         resolved.sort(key=lambda spec: self.SCOPES.index(spec.scope))
 165         return resolved
 166
 167
 168 class _ExecutionScope(object):
 169     '''Store execution/teardown state for a scope.'''
 170
 171     def __init__(self, scope, parent):
 172         self.scope = scope
 173         self.parent = parent
 174         self.cache = {}
 175         self.finalizers = []
 176
 177     def _find_scope(self, scope):
 178         context = self
 179         while context.scope != scope:
 180             context = context.parent
 181         return context
 182
 183     def execute(self, spec, test_fn):
 184         '''Execute a fixture and cache the result.'''
 185         context = self._find_scope(spec.scope)
 186         if spec.name in context.cache:
 187             return
 188         try:
 189             value, cleanup = self._execute_one(spec, test_fn)
 190             exc = None
 191         except Exception:
 192             value, cleanup, exc = None, None, sys.exc_info()[1]
 193         context.cache[spec.name] = value, exc
 194         if cleanup:
 195             context.finalizers.append(cleanup)
 196         if exc:
 197             raise exc
 198
 199     def cached_result(self, spec):
 200         '''Obtain the cached result for a previously executed fixture.'''
 201         entry = self._find_scope(spec.scope).cache.get(spec.name)
 202         if not entry:
 203             return None, False
 204         value, exc = entry
 205         if exc:
 206             raise exc
 207         return value, True
 208
 209     def _execute_one(self, spec, test_fn):
 210         # A fixture can only execute in the same or earlier scopes
 211         context_scope_index = _FixturesManager.SCOPES.index(self.scope)
 212         fixture_scope_index = _FixturesManager.SCOPES.index(spec.scope)
 213         assert fixture_scope_index <= context_scope_index
 214         if spec.params:
 215             # Do not invoke destroy, it is taken care of by the main request.
 216             subrequest = _FixtureRequest(self)
 217             subrequest.function = test_fn
 218             subrequest.fillfixtures(spec.params)
 219             fixtures = (subrequest.getfixturevalue(n) for n in spec.params)
 220             value = spec.func(*fixtures)  # Execute fixture
 221         else:
 222             value = spec.func()  # Execute fixture
 223         if not inspect.isgenerator(value):
 224             return value, None
 225
 226         @functools.wraps(value)
 227         def cleanup():
 228             try:
 229                 next(value)
 230             except StopIteration:
 231                 pass
 232             else:
 233                 raise RuntimeError('%s yielded more than once!' % (spec.name,))
 234         return next(value), cleanup
 235
 236     def destroy(self):
 237         exceptions = []
 238         for cleanup in self.finalizers:
 239             try:
 240                 cleanup()
 241             except:
 242                 exceptions.append(sys.exc_info()[1])
 243         self.cache.clear()
 244         self.finalizers.clear()
 245         if exceptions:
 246             raise exceptions[0]
 247
 248
 249 class _FixtureRequest(object):
 250     '''
 251     Holds state during a single test execution. See
 252     https://docs.pytest.org/en/latest/reference.html#request
 253     '''
 254
 255     def __init__(self, context):
 256         self._context = context
 257         self._fixtures_prepend = []  # fixtures added via usefixtures
 258         # XXX is there any need for .module or .cls?
 259         self.function = None  # test function, set before execution.
 260
 261     def fillfixtures(self, params):
 262         params = self._fixtures_prepend + params
 263         specs = _fallback.resolve_fixtures(params)
 264         for spec in specs:
 265             self._context.execute(spec, self.function)
 266
 267     def getfixturevalue(self, argname):
 268         spec = _fallback.lookup(argname)
 269         if not spec:
 270             assert argname == 'request'
 271             return self
 272         value, ok = self._context.cached_result(spec)
 273         if not ok:
 274             # If getfixturevalue is called directly from a setUp function, the
 275             # fixture value might not have computed before, so evaluate it now.
 276             # As the test function is not available, use None.
 277             self._context.execute(spec, test_fn=None)
 278             value, ok = self._context.cached_result(spec)
 279             assert ok, 'Failed to execute fixture %s' % (spec,)
 280         return value
 281
 282     def destroy(self):
 283         self._context.destroy()
 284
 285     def addfinalizer(self, finalizer):
 286         self._context.finalizers.append(finalizer)
 287
 288     @property
 289     def instance(self):
 290         return self.function.__self__
 291
 292     @property
 293     def config(self):
 294         '''The pytest config object associated with this request.'''
 295         return _config
 296
 297
 298 def _patch_unittest_testcase_class(cls):
 299     '''
 300     Patch the setUp and tearDown methods of the unittest.TestCase such that the
 301     fixtures are properly setup and destroyed.
 302     '''
 303
 304     def setUp(self):
 305         assert _session_context, 'must call create_session() first!'
 306         function_context = _ExecutionScope('function', _session_context)
 307         req = _FixtureRequest(function_context)
 308         req._fixtures_prepend = getattr(self, '_fixtures_prepend', [])
 309         self._fixture_request = req
 310         self._orig_setUp()
 311
 312     def tearDown(self):
 313         try:
 314             self._orig_tearDown()
 315         finally:
 316             self._fixture_request.destroy()
 317     # Only the leaf test case class should be decorated!
 318     assert not hasattr(cls, '_orig_setUp')
 319     assert not hasattr(cls, '_orig_tearDown')
 320     cls._orig_setUp, cls.setUp = cls.setUp, setUp
 321     cls._orig_tearDown, cls.tearDown = cls.tearDown, tearDown
 322
 323
 324 class _Config(object):
 325     def __init__(self, args):
 326         assert isinstance(args, argparse.Namespace)
 327         self.args = args
 328
 329     def getoption(self, name, default):
 330         '''Partial emulation for pytest Config.getoption.'''
 331         name = name.lstrip('-').replace('-', '_')
 332         return getattr(self.args, name, default)
 333
 334
 335 _fallback = None
 336 _session_context = None
 337 _config = None
 338
 339
 340 def init_fallback_fixtures_once():
 341     global _fallback
 342     assert not _use_native_pytest
 343     if _fallback:
 344         return
 345     _fallback = _FixturesManager()
 346     # Register standard fixtures here as needed
 347
 348
 349 def create_session(args=None):
 350     '''Start a test session where args is from argparse.'''
 351     global _session_context, _config
 352     assert not _use_native_pytest
 353     _session_context = _ExecutionScope('session', None)
 354     if args is None:
 355         args = argparse.Namespace()
 356     _config = _Config(args)
 357
 358
 359 def destroy_session():
 360     global _session_context
 361     assert not _use_native_pytest
 362     _session_context = None
 363
 364
 365 def skip(msg):
 366     '''Skip the executing test with the given message.'''
 367     if _use_native_pytest:
 368         pytest.skip(msg)
 369     else:
 370         raise unittest.SkipTest(msg)