""" core implementation of testing process: init, session, runtest loop. """ from __future__ import absolute_import, division, print_function import functools import os import sys import _pytest import _pytest._code import py try: from collections import MutableMapping as MappingMixin except ImportError: from UserDict import DictMixin as MappingMixin from _pytest.config import directory_arg, UsageError, hookimpl from _pytest.runner import collect_one_node, exit tracebackcutdir = py.path.local(_pytest.__file__).dirpath() # exitcodes for the command line EXIT_OK = 0 EXIT_TESTSFAILED = 1 EXIT_INTERRUPTED = 2 EXIT_INTERNALERROR = 3 EXIT_USAGEERROR = 4 EXIT_NOTESTSCOLLECTED = 5 def pytest_addoption(parser): parser.addini("norecursedirs", "directory patterns to avoid for recursion", type="args", default=['.*', 'build', 'dist', 'CVS', '_darcs', '{arch}', '*.egg', 'venv']) parser.addini("testpaths", "directories to search for tests when no files or directories are given in the command line.", type="args", default=[]) #parser.addini("dirpatterns", # "patterns specifying possible locations of test files", # type="linelist", default=["**/test_*.txt", # "**/test_*.py", "**/*_test.py"] #) group = parser.getgroup("general", "running and selection options") group._addoption('-x', '--exitfirst', action="store_const", dest="maxfail", const=1, help="exit instantly on first error or failed test."), group._addoption('--maxfail', metavar="num", action="store", type=int, dest="maxfail", default=0, help="exit after first num failures or errors.") group._addoption('--strict', action="store_true", help="run pytest in strict mode, warnings become errors.") group._addoption("-c", metavar="file", type=str, dest="inifilename", help="load configuration from `file` instead of trying to locate one of the implicit configuration files.") group._addoption("--continue-on-collection-errors", action="store_true", default=False, dest="continue_on_collection_errors", help="Force test execution even if collection errors occur.") group = parser.getgroup("collect", "collection") group.addoption('--collectonly', '--collect-only', action="store_true", help="only collect tests, don't execute them."), group.addoption('--pyargs', action="store_true", help="try to interpret all arguments as python packages.") group.addoption("--ignore", action="append", metavar="path", help="ignore path during collection (multi-allowed).") # when changing this to --conf-cut-dir, config.py Conftest.setinitial # needs upgrading as well group.addoption('--confcutdir', dest="confcutdir", default=None, metavar="dir", type=functools.partial(directory_arg, optname="--confcutdir"), help="only load conftest.py's relative to specified dir.") group.addoption('--noconftest', action="store_true", dest="noconftest", default=False, help="Don't load any conftest.py files.") group.addoption('--keepduplicates', '--keep-duplicates', action="store_true", dest="keepduplicates", default=False, help="Keep duplicate tests.") group = parser.getgroup("debugconfig", "test session debugging and configuration") group.addoption('--basetemp', dest="basetemp", default=None, metavar="dir", help="base temporary directory for this test run.") def pytest_namespace(): """keeping this one works around a deeper startup issue in pytest i tried to find it for a while but the amount of time turned unsustainable, so i put a hack in to revisit later """ return {} def pytest_configure(config): __import__('pytest').config = config # compatibiltiy def wrap_session(config, doit): """Skeleton command line program""" session = Session(config) session.exitstatus = EXIT_OK initstate = 0 try: try: config._do_configure() initstate = 1 config.hook.pytest_sessionstart(session=session) initstate = 2 session.exitstatus = doit(config, session) or 0 except UsageError: raise except KeyboardInterrupt: excinfo = _pytest._code.ExceptionInfo() if initstate < 2 and isinstance(excinfo.value, exit.Exception): sys.stderr.write('{0}: {1}\n'.format( excinfo.typename, excinfo.value.msg)) config.hook.pytest_keyboard_interrupt(excinfo=excinfo) session.exitstatus = EXIT_INTERRUPTED except: excinfo = _pytest._code.ExceptionInfo() config.notify_exception(excinfo, config.option) session.exitstatus = EXIT_INTERNALERROR if excinfo.errisinstance(SystemExit): sys.stderr.write("mainloop: caught Spurious SystemExit!\n") finally: excinfo = None # Explicitly break reference cycle. session.startdir.chdir() if initstate >= 2: config.hook.pytest_sessionfinish( session=session, exitstatus=session.exitstatus) config._ensure_unconfigure() return session.exitstatus def pytest_cmdline_main(config): return wrap_session(config, _main) def _main(config, session): """ default command line protocol for initialization, session, running tests and reporting. """ config.hook.pytest_collection(session=session) config.hook.pytest_runtestloop(session=session) if session.testsfailed: return EXIT_TESTSFAILED elif session.testscollected == 0: return EXIT_NOTESTSCOLLECTED def pytest_collection(session): return session.perform_collect() def pytest_runtestloop(session): if (session.testsfailed and not session.config.option.continue_on_collection_errors): raise session.Interrupted( "%d errors during collection" % session.testsfailed) if session.config.option.collectonly: return True for i, item in enumerate(session.items): nextitem = session.items[i+1] if i+1 < len(session.items) else None item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem) if session.shouldstop: raise session.Interrupted(session.shouldstop) return True def pytest_ignore_collect(path, config): p = path.dirpath() ignore_paths = config._getconftest_pathlist("collect_ignore", path=p) ignore_paths = ignore_paths or [] excludeopt = config.getoption("ignore") if excludeopt: ignore_paths.extend([py.path.local(x) for x in excludeopt]) if path in ignore_paths: return True # Skip duplicate paths. keepduplicates = config.getoption("keepduplicates") duplicate_paths = config.pluginmanager._duplicatepaths if not keepduplicates: if path in duplicate_paths: return True else: duplicate_paths.add(path) return False class FSHookProxy: def __init__(self, fspath, pm, remove_mods): self.fspath = fspath self.pm = pm self.remove_mods = remove_mods def __getattr__(self, name): x = self.pm.subset_hook_caller(name, remove_plugins=self.remove_mods) self.__dict__[name] = x return x class _CompatProperty(object): def __init__(self, name): self.name = name def __get__(self, obj, owner): if obj is None: return self # TODO: reenable in the features branch # warnings.warn( # "usage of {owner!r}.{name} is deprecated, please use pytest.{name} instead".format( # name=self.name, owner=type(owner).__name__), # PendingDeprecationWarning, stacklevel=2) return getattr(__import__('pytest'), self.name) class NodeKeywords(MappingMixin): def __init__(self, node): self.node = node self.parent = node.parent self._markers = {node.name: True} def __getitem__(self, key): try: return self._markers[key] except KeyError: if self.parent is None: raise return self.parent.keywords[key] def __setitem__(self, key, value): self._markers[key] = value def __delitem__(self, key): raise ValueError("cannot delete key in keywords dict") def __iter__(self): seen = set(self._markers) if self.parent is not None: seen.update(self.parent.keywords) return iter(seen) def __len__(self): return len(self.__iter__()) def keys(self): return list(self) def __repr__(self): return "" % (self.node, ) class Node(object): """ base class for Collector and Item the test collection tree. Collector subclasses have children, Items are terminal nodes.""" def __init__(self, name, parent=None, config=None, session=None): #: a unique name within the scope of the parent node self.name = name #: the parent collector node. self.parent = parent #: the pytest config object self.config = config or parent.config #: the session this node is part of self.session = session or parent.session #: filesystem path where this node was collected from (can be None) self.fspath = getattr(parent, 'fspath', None) #: keywords/markers collected from all scopes self.keywords = NodeKeywords(self) #: allow adding of extra keywords to use for matching self.extra_keyword_matches = set() # used for storing artificial fixturedefs for direct parametrization self._name2pseudofixturedef = {} @property def ihook(self): """ fspath sensitive hook proxy used to call pytest hooks""" return self.session.gethookproxy(self.fspath) Module = _CompatProperty("Module") Class = _CompatProperty("Class") Instance = _CompatProperty("Instance") Function = _CompatProperty("Function") File = _CompatProperty("File") Item = _CompatProperty("Item") def _getcustomclass(self, name): maybe_compatprop = getattr(type(self), name) if isinstance(maybe_compatprop, _CompatProperty): return getattr(__import__('pytest'), name) else: cls = getattr(self, name) # TODO: reenable in the features branch # warnings.warn("use of node.%s is deprecated, " # "use pytest_pycollect_makeitem(...) to create custom " # "collection nodes" % name, category=DeprecationWarning) return cls def __repr__(self): return "<%s %r>" %(self.__class__.__name__, getattr(self, 'name', None)) def warn(self, code, message): """ generate a warning with the given code and message for this item. """ assert isinstance(code, str) fslocation = getattr(self, "location", None) if fslocation is None: fslocation = getattr(self, "fspath", None) self.ihook.pytest_logwarning.call_historic(kwargs=dict( code=code, message=message, nodeid=self.nodeid, fslocation=fslocation)) # methods for ordering nodes @property def nodeid(self): """ a ::-separated string denoting its collection tree address. """ try: return self._nodeid except AttributeError: self._nodeid = x = self._makeid() return x def _makeid(self): return self.parent.nodeid + "::" + self.name def __hash__(self): return hash(self.nodeid) def setup(self): pass def teardown(self): pass def _memoizedcall(self, attrname, function): exattrname = "_ex_" + attrname failure = getattr(self, exattrname, None) if failure is not None: py.builtin._reraise(failure[0], failure[1], failure[2]) if hasattr(self, attrname): return getattr(self, attrname) try: res = function() except py.builtin._sysex: raise except: failure = sys.exc_info() setattr(self, exattrname, failure) raise setattr(self, attrname, res) return res def listchain(self): """ return list of all parent collectors up to self, starting from root of collection tree. """ chain = [] item = self while item is not None: chain.append(item) item = item.parent chain.reverse() return chain def add_marker(self, marker): """ dynamically add a marker object to the node. ``marker`` can be a string or pytest.mark.* instance. """ from _pytest.mark import MarkDecorator, MARK_GEN if isinstance(marker, py.builtin._basestring): marker = getattr(MARK_GEN, marker) elif not isinstance(marker, MarkDecorator): raise ValueError("is not a string or pytest.mark.* Marker") self.keywords[marker.name] = marker def get_marker(self, name): """ get a marker object from this node or None if the node doesn't have a marker with that name. """ val = self.keywords.get(name, None) if val is not None: from _pytest.mark import MarkInfo, MarkDecorator if isinstance(val, (MarkDecorator, MarkInfo)): return val def listextrakeywords(self): """ Return a set of all extra keywords in self and any parents.""" extra_keywords = set() item = self for item in self.listchain(): extra_keywords.update(item.extra_keyword_matches) return extra_keywords def listnames(self): return [x.name for x in self.listchain()] def addfinalizer(self, fin): """ register a function to be called when this node is finalized. This method can only be called when this node is active in a setup chain, for example during self.setup(). """ self.session._setupstate.addfinalizer(fin, self) def getparent(self, cls): """ get the next parent node (including ourself) which is an instance of the given class""" current = self while current and not isinstance(current, cls): current = current.parent return current def _prunetraceback(self, excinfo): pass def _repr_failure_py(self, excinfo, style=None): fm = self.session._fixturemanager if excinfo.errisinstance(fm.FixtureLookupError): return excinfo.value.formatrepr() tbfilter = True if self.config.option.fulltrace: style="long" else: tb = _pytest._code.Traceback([excinfo.traceback[-1]]) self._prunetraceback(excinfo) if len(excinfo.traceback) == 0: excinfo.traceback = tb tbfilter = False # prunetraceback already does it if style == "auto": style = "long" # XXX should excinfo.getrepr record all data and toterminal() process it? if style is None: if self.config.option.tbstyle == "short": style = "short" else: style = "long" try: os.getcwd() abspath = False except OSError: abspath = True return excinfo.getrepr(funcargs=True, abspath=abspath, showlocals=self.config.option.showlocals, style=style, tbfilter=tbfilter) repr_failure = _repr_failure_py class Collector(Node): """ Collector instances create children through collect() and thus iteratively build a tree. """ class CollectError(Exception): """ an error during collection, contains a custom message. """ def collect(self): """ returns a list of children (items and collectors) for this collection node. """ raise NotImplementedError("abstract") def repr_failure(self, excinfo): """ represent a collection failure. """ if excinfo.errisinstance(self.CollectError): exc = excinfo.value return str(exc.args[0]) return self._repr_failure_py(excinfo, style="short") def _prunetraceback(self, excinfo): if hasattr(self, 'fspath'): traceback = excinfo.traceback ntraceback = traceback.cut(path=self.fspath) if ntraceback == traceback: ntraceback = ntraceback.cut(excludepath=tracebackcutdir) excinfo.traceback = ntraceback.filter() class FSCollector(Collector): def __init__(self, fspath, parent=None, config=None, session=None): fspath = py.path.local(fspath) # xxx only for test_resultlog.py? name = fspath.basename if parent is not None: rel = fspath.relto(parent.fspath) if rel: name = rel name = name.replace(os.sep, "/") super(FSCollector, self).__init__(name, parent, config, session) self.fspath = fspath def _makeid(self): relpath = self.fspath.relto(self.config.rootdir) if os.sep != "/": relpath = relpath.replace(os.sep, "/") return relpath class File(FSCollector): """ base class for collecting tests from a file. """ class Item(Node): """ a basic test invocation item. Note that for a single function there might be multiple test invocation items. """ nextitem = None def __init__(self, name, parent=None, config=None, session=None): super(Item, self).__init__(name, parent, config, session) self._report_sections = [] def add_report_section(self, when, key, content): if content: self._report_sections.append((when, key, content)) def reportinfo(self): return self.fspath, None, "" @property def location(self): try: return self._location except AttributeError: location = self.reportinfo() # bestrelpath is a quite slow function cache = self.config.__dict__.setdefault("_bestrelpathcache", {}) try: fspath = cache[location[0]] except KeyError: fspath = self.session.fspath.bestrelpath(location[0]) cache[location[0]] = fspath location = (fspath, location[1], str(location[2])) self._location = location return location class NoMatch(Exception): """ raised if matching cannot locate a matching names. """ class Interrupted(KeyboardInterrupt): """ signals an interrupted test run. """ __module__ = 'builtins' # for py3 class Session(FSCollector): Interrupted = Interrupted def __init__(self, config): FSCollector.__init__(self, config.rootdir, parent=None, config=config, session=self) self.testsfailed = 0 self.testscollected = 0 self.shouldstop = False self.trace = config.trace.root.get("collection") self._norecursepatterns = config.getini("norecursedirs") self.startdir = py.path.local() self.config.pluginmanager.register(self, name="session") def _makeid(self): return "" @hookimpl(tryfirst=True) def pytest_collectstart(self): if self.shouldstop: raise self.Interrupted(self.shouldstop) @hookimpl(tryfirst=True) def pytest_runtest_logreport(self, report): if report.failed and not hasattr(report, 'wasxfail'): self.testsfailed += 1 maxfail = self.config.getvalue("maxfail") if maxfail and self.testsfailed >= maxfail: self.shouldstop = "stopping after %d failures" % ( self.testsfailed) pytest_collectreport = pytest_runtest_logreport def isinitpath(self, path): return path in self._initialpaths def gethookproxy(self, fspath): # check if we have the common case of running # hooks with all conftest.py filesall conftest.py pm = self.config.pluginmanager my_conftestmodules = pm._getconftestmodules(fspath) remove_mods = pm._conftest_plugins.difference(my_conftestmodules) if remove_mods: # one or more conftests are not in use at this fspath proxy = FSHookProxy(fspath, pm, remove_mods) else: # all plugis are active for this fspath proxy = self.config.hook return proxy def perform_collect(self, args=None, genitems=True): hook = self.config.hook try: items = self._perform_collect(args, genitems) self.config.pluginmanager.check_pending() hook.pytest_collection_modifyitems(session=self, config=self.config, items=items) finally: hook.pytest_collection_finish(session=self) self.testscollected = len(items) return items def _perform_collect(self, args, genitems): if args is None: args = self.config.args self.trace("perform_collect", self, args) self.trace.root.indent += 1 self._notfound = [] self._initialpaths = set() self._initialparts = [] self.items = items = [] for arg in args: parts = self._parsearg(arg) self._initialparts.append(parts) self._initialpaths.add(parts[0]) rep = collect_one_node(self) self.ihook.pytest_collectreport(report=rep) self.trace.root.indent -= 1 if self._notfound: errors = [] for arg, exc in self._notfound: line = "(no name %r in any of %r)" % (arg, exc.args[0]) errors.append("not found: %s\n%s" % (arg, line)) # XXX: test this raise UsageError(*errors) if not genitems: return rep.result else: if rep.passed: for node in rep.result: self.items.extend(self.genitems(node)) return items def collect(self): for parts in self._initialparts: arg = "::".join(map(str, parts)) self.trace("processing argument", arg) self.trace.root.indent += 1 try: for x in self._collect(arg): yield x except NoMatch: # we are inside a make_report hook so # we cannot directly pass through the exception self._notfound.append((arg, sys.exc_info()[1])) self.trace.root.indent -= 1 def _collect(self, arg): names = self._parsearg(arg) path = names.pop(0) if path.check(dir=1): assert not names, "invalid arg %r" % (arg,) for path in path.visit(fil=lambda x: x.check(file=1), rec=self._recurse, bf=True, sort=True): for x in self._collectfile(path): yield x else: assert path.check(file=1) for x in self.matchnodes(self._collectfile(path), names): yield x def _collectfile(self, path): ihook = self.gethookproxy(path) if not self.isinitpath(path): if ihook.pytest_ignore_collect(path=path, config=self.config): return () return ihook.pytest_collect_file(path=path, parent=self) def _recurse(self, path): ihook = self.gethookproxy(path.dirpath()) if ihook.pytest_ignore_collect(path=path, config=self.config): return for pat in self._norecursepatterns: if path.check(fnmatch=pat): return False ihook = self.gethookproxy(path) ihook.pytest_collect_directory(path=path, parent=self) return True def _tryconvertpyarg(self, x): """Convert a dotted module name to path. """ import pkgutil try: loader = pkgutil.find_loader(x) except ImportError: return x if loader is None: return x # This method is sometimes invoked when AssertionRewritingHook, which # does not define a get_filename method, is already in place: try: path = loader.get_filename(x) except AttributeError: # Retrieve path from AssertionRewritingHook: path = loader.modules[x][0].co_filename if loader.is_package(x): path = os.path.dirname(path) return path def _parsearg(self, arg): """ return (fspath, names) tuple after checking the file exists. """ parts = str(arg).split("::") if self.config.option.pyargs: parts[0] = self._tryconvertpyarg(parts[0]) relpath = parts[0].replace("/", os.sep) path = self.config.invocation_dir.join(relpath, abs=True) if not path.check(): if self.config.option.pyargs: raise UsageError( "file or package not found: " + arg + " (missing __init__.py?)") else: raise UsageError("file not found: " + arg) parts[0] = path return parts def matchnodes(self, matching, names): self.trace("matchnodes", matching, names) self.trace.root.indent += 1 nodes = self._matchnodes(matching, names) num = len(nodes) self.trace("matchnodes finished -> ", num, "nodes") self.trace.root.indent -= 1 if num == 0: raise NoMatch(matching, names[:1]) return nodes def _matchnodes(self, matching, names): if not matching or not names: return matching name = names[0] assert name nextnames = names[1:] resultnodes = [] for node in matching: if isinstance(node, Item): if not names: resultnodes.append(node) continue assert isinstance(node, Collector) rep = collect_one_node(node) if rep.passed: has_matched = False for x in rep.result: # TODO: remove parametrized workaround once collection structure contains parametrization if x.name == name or x.name.split("[")[0] == name: resultnodes.extend(self.matchnodes([x], nextnames)) has_matched = True # XXX accept IDs that don't have "()" for class instances if not has_matched and len(rep.result) == 1 and x.name == "()": nextnames.insert(0, name) resultnodes.extend(self.matchnodes([x], nextnames)) node.ihook.pytest_collectreport(report=rep) return resultnodes def genitems(self, node): self.trace("genitems", node) if isinstance(node, Item): node.ihook.pytest_itemcollected(item=node) yield node else: assert isinstance(node, Collector) rep = collect_one_node(node) if rep.passed: for subnode in rep.result: for x in self.genitems(subnode): yield x node.ihook.pytest_collectreport(report=rep)