import os
import warnings
from functools import lru_cache
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Union
import py
import _pytest._code
from _pytest._code.code import ExceptionChainRepr
from _pytest._code.code import ExceptionInfo
from _pytest._code.code import ReprExceptionInfo
from _pytest._code.source import getfslineno
from _pytest.compat import cached_property
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config import PytestPluginManager
from _pytest.deprecated import NODE_USE_FROM_PARENT
from _pytest.fixtures import FixtureDef
from _pytest.fixtures import FixtureLookupError
from _pytest.fixtures import FixtureLookupErrorRepr
from _pytest.mark.structures import Mark
from _pytest.mark.structures import MarkDecorator
from _pytest.mark.structures import NodeKeywords
from _pytest.outcomes import fail
from _pytest.outcomes import Failed
if TYPE_CHECKING:
# Imported here due to circular import.
from _pytest.main import Session # noqa: F401
from _pytest._code.code import _TracebackStyle # noqa: F401
SEP = "/"
tracebackcutdir = py.path.local(_pytest.__file__).dirpath()
[docs]@lru_cache(maxsize=None)
def _splitnode(nodeid):
"""Split a nodeid into constituent 'parts'.
Node IDs are strings, and can be things like:
''
'testing/code'
'testing/code/test_excinfo.py'
'testing/code/test_excinfo.py::TestFormattedExcinfo'
Return values are lists e.g.
[]
['testing', 'code']
['testing', 'code', 'test_excinfo.py']
['testing', 'code', 'test_excinfo.py', '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 'test_foo.py::Bar' with multiple elements 'test_foo.py', 'Bar'
parts[-1:] = parts[-1].split("::")
# Convert parts into a tuple to avoid possible errors with caching of a mutable type
return tuple(parts)
[docs]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(metaclass=NodeMeta):
""" base class for Collector and Item the test collection tree.
Collector subclasses have children, Items are terminal nodes."""
def __init__(
self,
name: str,
parent: Optional["Node"] = None,
config: Optional[Config] = None,
session: Optional["Session"] = None,
fspath: Optional[py.path.local] = None,
nodeid: Optional[str] = None,
) -> 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
if config:
self.config = config
else:
if not parent:
raise TypeError("config or parent must be provided")
self.config = parent.config
#: the session this node is part of
if session:
self.session = session
else:
if not parent:
raise TypeError("session or parent must be provided")
self.session = parent.session
#: filesystem path where this node was collected from (can be None)
self.fspath = fspath or getattr(parent, "fspath", None) # type: py.path.local
#: 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:
if not self.parent:
raise TypeError("nodeid or parent must be provided")
self._nodeid = self.parent.nodeid
if self.name != "()":
self._nodeid += "::" + self.name
self._chain = self.listchain()
[docs] @classmethod
def from_parent(cls, parent: "Node", **kw):
"""
Public Constructor for Nodes
This indirection got introduced in order to enable removing
the fragile logic from the node constructors.
Subclasses can use ``super().from_parent(...)`` when overriding the construction
:param parent: the parent node of this test Node
"""
if "config" in kw:
raise TypeError("config is not a valid argument for from_parent")
if "session" in kw:
raise TypeError("session is not a valid argument for from_parent")
return cls._create(parent=parent, **kw)
@property
def ihook(self):
""" fspath sensitive hook proxy used to call pytest hooks"""
return self.session.gethookproxy(self.fspath)
def __repr__(self):
return "<{} nodeid={!r}>".format(
self.__class__.__name__, getattr(self, "nodeid", None)
)
def __str__(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)
warnings.warn_explicit(
warning,
category=None,
filename=str(path),
lineno=lineno + 1 if lineno is not None else None,
)
# methods for ordering nodes
@property
def nodeid(self) -> str:
""" a ::-separated string denoting its collection tree address. """
return self._nodeid
def __hash__(self):
return hash(self.nodeid)
[docs] def teardown(self):
pass
[docs] def listchain(self) -> List["Node"]:
""" return list of all parent collectors up to self,
starting from root of collection tree. """
if not self.parent:
return [self]
return self.parent.listchain() + [self]
[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_.name] = 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._chain):
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 listnames(self):
return [x.name for x in self._chain]
[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 # type: Optional[Node]
while current and not isinstance(current, cls):
current = current.parent
return current
[docs] def _prunetraceback(self, excinfo):
pass
[docs] def _repr_failure_py(
self,
excinfo: ExceptionInfo[Union[Failed, FixtureLookupError]],
default_style: "_TracebackStyle" = None,
) -> Union[str, ReprExceptionInfo, ExceptionChainRepr, FixtureLookupErrorRepr]:
fulltrace = self.config.getoption("fulltrace", False)
if (
not fulltrace
and isinstance(excinfo.value, fail.Exception)
and not excinfo.value.pytrace
):
return str(excinfo.value)
if isinstance(excinfo.value, FixtureLookupError):
return excinfo.value.formatrepr()
# XXX should excinfo.getrepr record all data and toterminal() process it?
# XXX: does not distinguish between default/unset and --tb=auto.
tbstyle = self.config.getoption("tbstyle", "auto")
if fulltrace:
style = "long" if tbstyle == "auto" else tbstyle # type: _TracebackStyle
else:
if tbstyle == "auto":
if default_style is None:
style = "long"
else:
style = default_style
else:
style = tbstyle
self._prunetraceback(excinfo)
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,
)
[docs] def repr_failure(
self, excinfo, style=None
) -> Union[str, ReprExceptionInfo, ExceptionChainRepr, FixtureLookupErrorRepr]:
return self._repr_failure_py(excinfo, style)
[docs]def get_fslocation_from_item(
item: "Item",
) -> Tuple[Union[str, py.path.local], Optional[int]]:
"""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.
"""
try:
return item.location[:2]
except AttributeError:
pass
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) and not self.config.getoption(
"fulltrace", False
):
exc = excinfo.value
return str(exc.args[0])
return self._repr_failure_py(excinfo, default_style="short")
[docs] 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)
ntraceback = ntraceback.filter()
if ntraceback:
excinfo.traceback = ntraceback
[docs]class FSHookProxy:
def __init__(
self, fspath: py.path.local, pm: PytestPluginManager, remove_mods
) -> None:
self.fspath = fspath
self.pm = pm
self.remove_mods = remove_mods
def __getattr__(self, name: str):
x = self.pm.subset_hook_caller(name, remove_plugins=self.remove_mods)
self.__dict__[name] = x
return x
[docs]class FSCollector(Collector):
def __init__(
self,
fspath: py.path.local,
parent=None,
config=None,
session=None,
nodeid: Optional[str] = None,
) -> None:
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 = session._node_location_to_relpath(self.fspath)
if os.sep != SEP:
nodeid = nodeid.replace(os.sep, SEP)
super().__init__(name, parent, config, session, nodeid=nodeid, fspath=fspath)
self._norecursepatterns = self.config.getini("norecursedirs")
[docs] @classmethod
def from_parent(cls, parent, *, fspath):
"""
The public constructor
"""
return super().from_parent(parent=parent, fspath=fspath)
[docs] def _gethookproxy(self, fspath: py.path.local):
# check if we have the common case of running
# hooks with all conftest.py files
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 plugins are active for this fspath
proxy = self.config.hook
return proxy
[docs] def _is_ignored(self, ihook, path: py.path.local) -> bool:
"""Wrap pytest_ignore_collect to support tuple return value."""
ret = ihook.pytest_ignore_collect(
path=path, config=self.config
) # type: Optional[Union[bool, Tuple[bool, Optional[str]]]]
if isinstance(ret, tuple):
return ret[0]
return bool(ret)
[docs] def _recurse(self, dirpath: py.path.local) -> bool:
if dirpath.basename == "__pycache__":
return False
if self._is_ignored(self._gethookproxy(dirpath.dirpath()), dirpath):
return False
for pat in self._norecursepatterns:
if dirpath.check(fnmatch=pat):
return False
ihook = self._gethookproxy(dirpath)
ihook.pytest_collect_directory(path=dirpath, parent=self)
return True
[docs] def _collectfile(self, path, handle_dupes=True):
assert (
path.isfile()
), "{!r} is not a file (isdir={!r}, exists={!r}, islink={!r})".format(
path, path.isdir(), path.exists(), path.islink()
)
ihook = self.gethookproxy(path)
if not self.isinitpath(path) and self._is_ignored(ihook, path):
return ()
if handle_dupes:
keepduplicates = self.config.getoption("keepduplicates")
if not keepduplicates:
duplicate_paths = self.config.pluginmanager._duplicatepaths
if path in duplicate_paths:
return ()
else:
duplicate_paths.add(path)
return ihook.pytest_collect_file(path=path, parent=self)
[docs]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 runtest(self) -> None:
raise NotImplementedError("runtest must be implemented by Item subclass")
[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))
[docs] def reportinfo(self) -> Tuple[Union[py.path.local, str], Optional[int], str]:
return self.fspath, None, ""
@cached_property
def location(self) -> Tuple[str, Optional[int], str]:
location = self.reportinfo()
if isinstance(location[0], py.path.local):
fspath = location[0]
else:
fspath = py.path.local(location[0])
relfspath = self.session._node_location_to_relpath(fspath)
assert type(location[2]) is str
return (relfspath, location[1], location[2])