Source code for _pytest.nodes

import os
import warnings
from functools import lru_cache
from typing import Any
from typing import Dict
from typing import List
from typing import Set
from typing import Tuple
from typing import Union

import py

import _pytest._code
from _pytest.compat import getfslineno
from _pytest.mark.structures import Mark
from _pytest.mark.structures import MarkDecorator
from _pytest.mark.structures import NodeKeywords
from _pytest.outcomes import fail

if False:  # TYPE_CHECKING
    # Imported here due to circular import.
    from _pytest.fixtures import FixtureDef

SEP = "/"

tracebackcutdir = py.path.local(_pytest.__file__).dirpath()

def _splitnode(nodeid):
    """Split a nodeid into constituent 'parts'.

    Node IDs are strings, and can be things like:

    Return values are lists e.g.
        ['testing', 'code']
        ['testing', 'code', '']
        ['testing', 'code', '', 'TestFormattedExcinfo', '()']
    if nodeid == "":
        # If there is no root node at all, return an empty list so the caller's logic can remain sane
        return ()
    parts = nodeid.split(SEP)
    # Replace single last element '' with multiple elements '', 'Bar'
    parts[-1:] = parts[-1].split("::")
    # Convert parts into a tuple to avoid possible errors with caching of a mutable type
    return tuple(parts)

def ischildnode(baseid, nodeid):
    """Return True if the nodeid is a child node of the baseid.

    E.g. 'foo/bar::Baz' is a child of 'foo', 'foo/bar' and 'foo/bar::Baz', but not of 'foo/blorp'
    base_parts = _splitnode(baseid)
    node_parts = _splitnode(nodeid)
    if len(node_parts) < len(base_parts):
        return False
    return node_parts[: len(base_parts)] == base_parts

[docs]class Node: """ 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, fspath=None, nodeid=None ): #: a unique name within the scope of the parent node = 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 = fspath or getattr(parent, "fspath", None) #: keywords/markers collected from all scopes self.keywords = NodeKeywords(self) #: the marker objects belonging to this node self.own_markers = [] # type: List[Mark] #: allow adding of extra keywords to use for matching self.extra_keyword_matches = set() # type: Set[str] # used for storing artificial fixturedefs for direct parametrization self._name2pseudofixturedef = {} # type: Dict[str, FixtureDef] if nodeid is not None: assert "::()" not in nodeid self._nodeid = nodeid else: self._nodeid = self.parent.nodeid if != "()": self._nodeid += "::" + @property def ihook(self): """ fspath sensitive hook proxy used to call pytest hooks""" return self.session.gethookproxy(self.fspath) def __repr__(self): return "<{} {}>".format(self.__class__.__name__, getattr(self, "name", None))
[docs] def warn(self, warning): """Issue a warning for this item. Warnings will be displayed after the test session, unless explicitly suppressed :param Warning warning: the warning instance to issue. Must be a subclass of PytestWarning. :raise ValueError: if ``warning`` instance is not a subclass of PytestWarning. Example usage: .. code-block:: python node.warn(PytestWarning("some message")) """ from _pytest.warning_types import PytestWarning if not isinstance(warning, PytestWarning): raise ValueError( "warning must be an instance of PytestWarning or subclass, got {!r}".format( warning ) ) path, lineno = get_fslocation_from_item(self) # Type ignored: warnings.warn_explicit( # type: ignore warning, category=None, filename=str(path), lineno=lineno + 1 if lineno is not None else None, )
# methods for ordering nodes @property def nodeid(self): """ a ::-separated string denoting its collection tree address. """ return self._nodeid def __hash__(self): return hash(self.nodeid) def setup(self): pass def teardown(self): pass
[docs] 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
[docs] def add_marker( self, marker: Union[str, MarkDecorator], append: bool = True ) -> None: """dynamically add a marker object to the node. :type marker: ``str`` or ``pytest.mark.*`` object :param marker: ``append=True`` whether to append the marker, if ``False`` insert at position ``0``. """ from _pytest.mark import MARK_GEN if isinstance(marker, MarkDecorator): marker_ = marker elif isinstance(marker, str): marker_ = getattr(MARK_GEN, marker) else: raise ValueError("is not a string or pytest.mark.* Marker") self.keywords[] = marker if append: self.own_markers.append(marker_.mark) else: self.own_markers.insert(0, marker_.mark)
[docs] def iter_markers(self, name=None): """ :param name: if given, filter the results by the name attribute iterate over all markers of the node """ return (x[1] for x in self.iter_markers_with_node(name=name))
[docs] def iter_markers_with_node(self, name=None): """ :param name: if given, filter the results by the name attribute iterate over all markers of the node returns sequence of tuples (node, mark) """ for node in reversed(self.listchain()): for mark in node.own_markers: if name is None or getattr(mark, "name", None) == name: yield node, mark
[docs] def get_closest_marker(self, name, default=None): """return the first marker matching the name, from closest (for example function) to farther level (for example module level). :param default: fallback return value of no marker was found :param name: name to filter by """ return next(self.iter_markers(name=name), default)
[docs] def listextrakeywords(self): """ Return a set of all extra keywords in self and any parents.""" extra_keywords = set() # type: Set[str] for item in self.listchain(): extra_keywords.update(item.extra_keyword_matches) return extra_keywords
def listnames(self): return [ for x in self.listchain()]
[docs] 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)
[docs] 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): # Type ignored: see comment where fail.Exception is defined. if excinfo.errisinstance(fail.Exception): # type: ignore if not excinfo.value.pytrace: return str(excinfo.value) fm = self.session._fixturemanager if excinfo.errisinstance(fm.FixtureLookupError): return excinfo.value.formatrepr() if self.config.getoption("fulltrace", False): style = "long" else: tb = _pytest._code.Traceback([excinfo.traceback[-1]]) self._prunetraceback(excinfo) if len(excinfo.traceback) == 0: excinfo.traceback = tb if style == "auto": style = "long" # XXX should excinfo.getrepr record all data and toterminal() process it? if style is None: if self.config.getoption("tbstyle", "auto") == "short": style = "short" else: style = "long" if self.config.getoption("verbose", 0) > 1: truncate_locals = False else: truncate_locals = True try: os.getcwd() abspath = False except OSError: abspath = True return excinfo.getrepr( funcargs=True, abspath=abspath, showlocals=self.config.getoption("showlocals", False), style=style, tbfilter=False, # pruned already, or in --fulltrace mode. truncate_locals=truncate_locals, ) def repr_failure(self, excinfo, style=None): return self._repr_failure_py(excinfo, style)
def get_fslocation_from_item(item): """Tries to extract the actual location from an item, depending on available attributes: * "fslocation": a pair (path, lineno) * "obj": a Python object that the item wraps. * "fspath": just a path :rtype: a tuple of (str|LocalPath, int) with filename and line number. """ result = getattr(item, "location", None) if result is not None: return result[:2] obj = getattr(item, "obj", None) if obj is not None: return getfslineno(obj) return getattr(item, "fspath", "unknown location"), -1
[docs]class Collector(Node): """ Collector instances create children through collect() and thus iteratively build a tree. """
[docs] class CollectError(Exception): """ an error during collection, contains a custom message. """
[docs] def collect(self): """ returns a list of children (items and collectors) for this collection node. """ raise NotImplementedError("abstract")
[docs] def repr_failure(self, excinfo): """ represent a collection failure. """ if excinfo.errisinstance(self.CollectError): exc = excinfo.value return str(exc.args[0]) # Respect explicit tbstyle option, but default to "short" # (None._repr_failure_py defaults to "long" without "fulltrace" option). tbstyle = self.config.getoption("tbstyle", "auto") if tbstyle == "auto": tbstyle = "short" return self._repr_failure_py(excinfo, style=tbstyle)
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()
def _check_initialpaths_for_relpath(session, fspath): for initial_path in session._initialpaths: if fspath.common(initial_path) == initial_path: return fspath.relto(initial_path)
[docs]class FSCollector(Collector): def __init__(self, fspath, parent=None, config=None, session=None, nodeid=None): fspath = py.path.local(fspath) # xxx only for name = fspath.basename if parent is not None: rel = fspath.relto(parent.fspath) if rel: name = rel name = name.replace(os.sep, SEP) self.fspath = fspath session = session or parent.session if nodeid is None: nodeid = self.fspath.relto(session.config.rootdir) if not nodeid: nodeid = _check_initialpaths_for_relpath(session, fspath) if nodeid and os.sep != SEP: nodeid = nodeid.replace(os.sep, SEP) super().__init__(name, parent, config, session, nodeid=nodeid, fspath=fspath)
class File(FSCollector): """ base class for collecting tests from a file. """
[docs]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, nodeid=None): super().__init__(name, parent, config, session, nodeid=nodeid) self._report_sections = [] # type: List[Tuple[str, str, str]] #: user properties is a list of tuples (name, value) that holds user #: defined properties for this test. self.user_properties = [] # type: List[Tuple[str, Any]]
[docs] def add_report_section(self, when: str, key: str, content: str) -> None: """ Adds a new report section, similar to what's done internally to add stdout and stderr captured output:: item.add_report_section("call", "stdout", "report section contents") :param str when: One of the possible capture states, ``"setup"``, ``"call"``, ``"teardown"``. :param str key: Name of the section, can be customized at will. Pytest uses ``"stdout"`` and ``"stderr"`` internally. :param str content: The full contents as a string. """ 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() fspath = self.session._node_location_to_relpath(location[0]) location = (fspath, location[1], str(location[2])) self._location = location return location