diff options
Diffstat (limited to 'lib/spack/external/_pytest/runner.py')
| -rw-r--r-- | lib/spack/external/_pytest/runner.py | 224 |
1 files changed, 77 insertions, 147 deletions
diff --git a/lib/spack/external/_pytest/runner.py b/lib/spack/external/_pytest/runner.py index eb29e7370c..b643fa3c91 100644 --- a/lib/spack/external/_pytest/runner.py +++ b/lib/spack/external/_pytest/runner.py @@ -1,29 +1,26 @@ """ basic collect and runtest protocol implementations """ +from __future__ import absolute_import, division, print_function + import bdb +import os import sys from time import time import py -import pytest +from _pytest.compat import _PY2 from _pytest._code.code import TerminalRepr, ExceptionInfo - - -def pytest_namespace(): - return { - 'fail' : fail, - 'skip' : skip, - 'importorskip' : importorskip, - 'exit' : exit, - } +from _pytest.outcomes import skip, Skipped, TEST_OUTCOME # # pytest plugin hooks + def pytest_addoption(parser): group = parser.getgroup("terminal reporting", "reporting", after="general") group.addoption('--durations', - action="store", type=int, default=None, metavar="N", - help="show N slowest setup/test durations (N=0 for all)."), + action="store", type=int, default=None, metavar="N", + help="show N slowest setup/test durations (N=0 for all)."), + def pytest_terminal_summary(terminalreporter): durations = terminalreporter.config.option.durations @@ -48,16 +45,16 @@ def pytest_terminal_summary(terminalreporter): for rep in dlist: nodeid = rep.nodeid.replace("::()::", "::") tr.write_line("%02.2fs %-8s %s" % - (rep.duration, rep.when, nodeid)) + (rep.duration, rep.when, nodeid)) + def pytest_sessionstart(session): session._setupstate = SetupState() + + def pytest_sessionfinish(session): session._setupstate.teardown_all() -class NodeInfo: - def __init__(self, location): - self.location = location def pytest_runtest_protocol(item, nextitem): item.ihook.pytest_runtest_logstart( @@ -66,6 +63,7 @@ def pytest_runtest_protocol(item, nextitem): runtestprotocol(item, nextitem=nextitem) return True + def runtestprotocol(item, log=True, nextitem=None): hasrequest = hasattr(item, "_request") if hasrequest and not item._request: @@ -78,7 +76,7 @@ def runtestprotocol(item, log=True, nextitem=None): if not item.config.option.setuponly: reports.append(call_and_report(item, "call", log)) reports.append(call_and_report(item, "teardown", log, - nextitem=nextitem)) + nextitem=nextitem)) # after all teardown hooks have been called # want funcargs and request info to go away if hasrequest: @@ -86,6 +84,7 @@ def runtestprotocol(item, log=True, nextitem=None): item.funcargs = None return reports + def show_test_item(item): """Show test function, parameters and the fixtures of the test item.""" tw = item.config.get_terminal_writer() @@ -96,10 +95,14 @@ def show_test_item(item): if used_fixtures: tw.write(' (fixtures used: {0})'.format(', '.join(used_fixtures))) + def pytest_runtest_setup(item): + _update_current_test_var(item, 'setup') item.session._setupstate.prepare(item) + def pytest_runtest_call(item): + _update_current_test_var(item, 'call') try: item.runtest() except Exception: @@ -112,8 +115,29 @@ def pytest_runtest_call(item): del tb # Get rid of it in this namespace raise + def pytest_runtest_teardown(item, nextitem): + _update_current_test_var(item, 'teardown') item.session._setupstate.teardown_exact(item, nextitem) + _update_current_test_var(item, None) + + +def _update_current_test_var(item, when): + """ + Update PYTEST_CURRENT_TEST to reflect the current item and stage. + + If ``when`` is None, delete PYTEST_CURRENT_TEST from the environment. + """ + var_name = 'PYTEST_CURRENT_TEST' + if when: + value = '{0} ({1})'.format(item.nodeid, when) + if _PY2: + # python 2 doesn't like null bytes on environment variables (see #2644) + value = value.replace('\x00', '(null)') + os.environ[var_name] = value + else: + os.environ.pop(var_name) + def pytest_report_teststatus(report): if report.when in ("setup", "teardown"): @@ -139,21 +163,25 @@ def call_and_report(item, when, log=True, **kwds): hook.pytest_exception_interact(node=item, call=call, report=report) return report + def check_interactive_exception(call, report): return call.excinfo and not ( - hasattr(report, "wasxfail") or - call.excinfo.errisinstance(skip.Exception) or - call.excinfo.errisinstance(bdb.BdbQuit)) + hasattr(report, "wasxfail") or + call.excinfo.errisinstance(skip.Exception) or + call.excinfo.errisinstance(bdb.BdbQuit)) + def call_runtest_hook(item, when, **kwds): hookname = "pytest_runtest_" + when ihook = getattr(item.ihook, hookname) return CallInfo(lambda: ihook(item=item, **kwds), when=when) + class CallInfo: """ Result/Exception info a function invocation. """ #: None or ExceptionInfo object. excinfo = None + def __init__(self, func, when): #: context of invocation: one of "setup", "call", #: "teardown", "memocollect" @@ -164,7 +192,7 @@ class CallInfo: except KeyboardInterrupt: self.stop = time() raise - except: + except: # noqa self.excinfo = ExceptionInfo() self.stop = time() @@ -175,6 +203,7 @@ class CallInfo: status = "result: %r" % (self.result,) return "<CallInfo when=%r %s>" % (self.when, status) + def getslaveinfoline(node): try: return node._slaveinfocache @@ -185,6 +214,7 @@ def getslaveinfoline(node): d['id'], d['sysplatform'], ver, d['executable']) return s + class BaseReport(object): def __init__(self, **kw): @@ -249,10 +279,11 @@ class BaseReport(object): def fspath(self): return self.nodeid.split("::")[0] + def pytest_runtest_makereport(item, call): when = call.when - duration = call.stop-call.start - keywords = dict([(x,1) for x in item.keywords]) + duration = call.stop - call.start + keywords = dict([(x, 1) for x in item.keywords]) excinfo = call.excinfo sections = [] if not call.excinfo: @@ -262,7 +293,7 @@ def pytest_runtest_makereport(item, call): if not isinstance(excinfo, ExceptionInfo): outcome = "failed" longrepr = excinfo - elif excinfo.errisinstance(pytest.skip.Exception): + elif excinfo.errisinstance(skip.Exception): outcome = "skipped" r = excinfo._getreprcrash() longrepr = (str(r.path), r.lineno, r.message) @@ -270,19 +301,21 @@ def pytest_runtest_makereport(item, call): outcome = "failed" if call.when == "call": longrepr = item.repr_failure(excinfo) - else: # exception in setup or teardown + else: # exception in setup or teardown longrepr = item._repr_failure_py(excinfo, - style=item.config.option.tbstyle) + style=item.config.option.tbstyle) for rwhen, key, content in item._report_sections: - sections.append(("Captured %s %s" %(key, rwhen), content)) + sections.append(("Captured %s %s" % (key, rwhen), content)) return TestReport(item.nodeid, item.location, keywords, outcome, longrepr, when, sections, duration) + class TestReport(BaseReport): """ Basic test report object (also used for setup and teardown calls if they fail). """ + def __init__(self, nodeid, location, keywords, outcome, longrepr, when, sections=(), duration=0, **extra): #: normalized collection node id @@ -321,16 +354,21 @@ class TestReport(BaseReport): return "<TestReport %r when=%r outcome=%r>" % ( self.nodeid, self.when, self.outcome) + class TeardownErrorReport(BaseReport): outcome = "failed" when = "teardown" + def __init__(self, longrepr, **extra): self.longrepr = longrepr self.sections = [] self.__dict__.update(extra) + def pytest_make_collect_report(collector): - call = CallInfo(collector._memocollect, "memocollect") + call = CallInfo( + lambda: list(collector.collect()), + 'collect') longrepr = None if not call.excinfo: outcome = "passed" @@ -348,7 +386,7 @@ def pytest_make_collect_report(collector): errorinfo = CollectErrorRepr(errorinfo) longrepr = errorinfo rep = CollectReport(collector.nodeid, outcome, longrepr, - getattr(call, 'result', None)) + getattr(call, 'result', None)) rep.call = call # see collect_one_node return rep @@ -369,16 +407,20 @@ class CollectReport(BaseReport): def __repr__(self): return "<CollectReport %r lenresult=%s outcome=%r>" % ( - self.nodeid, len(self.result), self.outcome) + self.nodeid, len(self.result), self.outcome) + class CollectErrorRepr(TerminalRepr): def __init__(self, msg): self.longrepr = msg + def toterminal(self, out): out.line(self.longrepr, red=True) + class SetupState(object): """ shared state for setting up/tearing down test items or collectors. """ + def __init__(self): self.stack = [] self._finalizers = {} @@ -390,7 +432,7 @@ class SetupState(object): """ assert colitem and not isinstance(colitem, tuple) assert py.builtin.callable(finalizer) - #assert colitem in self.stack # some unit tests don't setup stack :/ + # assert colitem in self.stack # some unit tests don't setup stack :/ self._finalizers.setdefault(colitem, []).append(finalizer) def _pop_and_teardown(self): @@ -404,7 +446,7 @@ class SetupState(object): fin = finalizers.pop() try: fin() - except Exception: + except TEST_OUTCOME: # XXX Only first exception will be seen by user, # ideally all should be reported. if exc is None: @@ -418,7 +460,7 @@ class SetupState(object): colitem.teardown() for colitem in self._finalizers: assert colitem is None or colitem in self.stack \ - or isinstance(colitem, tuple) + or isinstance(colitem, tuple) def teardown_all(self): while self.stack: @@ -451,10 +493,11 @@ class SetupState(object): self.stack.append(col) try: col.setup() - except Exception: + except TEST_OUTCOME: col._prepare_exc = sys.exc_info() raise + def collect_one_node(collector): ihook = collector.ihook ihook.pytest_collectstart(collector=collector) @@ -463,116 +506,3 @@ def collect_one_node(collector): if call and check_interactive_exception(call, rep): ihook.pytest_exception_interact(node=collector, call=call, report=rep) return rep - - -# ============================================================= -# Test OutcomeExceptions and helpers for creating them. - - -class OutcomeException(Exception): - """ OutcomeException and its subclass instances indicate and - contain info about test and collection outcomes. - """ - def __init__(self, msg=None, pytrace=True): - Exception.__init__(self, msg) - self.msg = msg - self.pytrace = pytrace - - def __repr__(self): - if self.msg: - val = self.msg - if isinstance(val, bytes): - val = py._builtin._totext(val, errors='replace') - return val - return "<%s instance>" %(self.__class__.__name__,) - __str__ = __repr__ - -class Skipped(OutcomeException): - # XXX hackish: on 3k we fake to live in the builtins - # in order to have Skipped exception printing shorter/nicer - __module__ = 'builtins' - - def __init__(self, msg=None, pytrace=True, allow_module_level=False): - OutcomeException.__init__(self, msg=msg, pytrace=pytrace) - self.allow_module_level = allow_module_level - - -class Failed(OutcomeException): - """ raised from an explicit call to pytest.fail() """ - __module__ = 'builtins' - - -class Exit(KeyboardInterrupt): - """ raised for immediate program exits (no tracebacks/summaries)""" - def __init__(self, msg="unknown reason"): - self.msg = msg - KeyboardInterrupt.__init__(self, msg) - -# exposed helper methods - -def exit(msg): - """ exit testing process as if KeyboardInterrupt was triggered. """ - __tracebackhide__ = True - raise Exit(msg) - - -exit.Exception = Exit - - -def skip(msg=""): - """ skip an executing test with the given message. Note: it's usually - better to use the pytest.mark.skipif marker to declare a test to be - skipped under certain conditions like mismatching platforms or - dependencies. See the pytest_skipping plugin for details. - """ - __tracebackhide__ = True - raise Skipped(msg=msg) - - -skip.Exception = Skipped - - -def fail(msg="", pytrace=True): - """ explicitly fail an currently-executing test with the given Message. - - :arg pytrace: if false the msg represents the full failure information - and no python traceback will be reported. - """ - __tracebackhide__ = True - raise Failed(msg=msg, pytrace=pytrace) - - -fail.Exception = Failed - - -def importorskip(modname, minversion=None): - """ return imported module if it has at least "minversion" as its - __version__ attribute. If no minversion is specified the a skip - is only triggered if the module can not be imported. - """ - __tracebackhide__ = True - compile(modname, '', 'eval') # to catch syntaxerrors - should_skip = False - try: - __import__(modname) - except ImportError: - # Do not raise chained exception here(#1485) - should_skip = True - if should_skip: - raise Skipped("could not import %r" %(modname,), allow_module_level=True) - mod = sys.modules[modname] - if minversion is None: - return mod - verattr = getattr(mod, '__version__', None) - if minversion is not None: - try: - from pkg_resources import parse_version as pv - except ImportError: - raise Skipped("we have a required version for %r but can not import " - "pkg_resources to parse version strings." % (modname,), - allow_module_level=True) - if verattr is None or pv(verattr) < pv(minversion): - raise Skipped("module %r has __version__ %r, required is: %r" %( - modname, verattr, minversion), allow_module_level=True) - return mod - |