Type annotate fixtures.py & related

This commit is contained in:
Ran Benita 2020-05-01 14:40:15 +03:00
parent be00e12d47
commit ef34729541
6 changed files with 193 additions and 106 deletions

View File

@ -5,9 +5,19 @@ import sys
import warnings import warnings
from collections import defaultdict from collections import defaultdict
from collections import deque from collections import deque
from types import TracebackType
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict from typing import Dict
from typing import Iterable
from typing import Iterator
from typing import List from typing import List
from typing import Optional
from typing import Sequence
from typing import Set
from typing import Tuple from typing import Tuple
from typing import Union
import attr import attr
import py import py
@ -29,6 +39,8 @@ from _pytest.compat import NOTSET
from _pytest.compat import order_preserving_dict from _pytest.compat import order_preserving_dict
from _pytest.compat import safe_getattr from _pytest.compat import safe_getattr
from _pytest.compat import TYPE_CHECKING from _pytest.compat import TYPE_CHECKING
from _pytest.config import _PluggyPlugin
from _pytest.config import Config
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
from _pytest.deprecated import FILLFUNCARGS from _pytest.deprecated import FILLFUNCARGS
from _pytest.deprecated import FIXTURE_POSITIONAL_ARGUMENTS from _pytest.deprecated import FIXTURE_POSITIONAL_ARGUMENTS
@ -38,16 +50,31 @@ from _pytest.outcomes import fail
from _pytest.outcomes import TEST_OUTCOME from _pytest.outcomes import TEST_OUTCOME
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import NoReturn
from typing import Type from typing import Type
from typing_extensions import Literal
from _pytest import nodes from _pytest import nodes
from _pytest.main import Session from _pytest.main import Session
from _pytest.python import Metafunc
_Scope = Literal["session", "package", "module", "class", "function"]
_FixtureCachedResult = Tuple[
# The result.
Optional[object],
# Cache key.
object,
# Exc info if raised.
Optional[Tuple["Type[BaseException]", BaseException, TracebackType]],
]
@attr.s(frozen=True) @attr.s(frozen=True)
class PseudoFixtureDef: class PseudoFixtureDef:
cached_result = attr.ib() cached_result = attr.ib(type="_FixtureCachedResult")
scope = attr.ib() scope = attr.ib(type="_Scope")
def pytest_sessionstart(session: "Session") -> None: def pytest_sessionstart(session: "Session") -> None:
@ -92,7 +119,7 @@ def scopeproperty(name=None, doc=None):
return decoratescope return decoratescope
def get_scope_package(node, fixturedef): def get_scope_package(node, fixturedef: "FixtureDef"):
import pytest import pytest
cls = pytest.Package cls = pytest.Package
@ -114,7 +141,9 @@ def get_scope_node(node, scope):
return node.getparent(cls) return node.getparent(cls)
def add_funcarg_pseudo_fixture_def(collector, metafunc, fixturemanager): def add_funcarg_pseudo_fixture_def(
collector, metafunc: "Metafunc", fixturemanager: "FixtureManager"
) -> None:
# this function will transform all collected calls to a functions # this function will transform all collected calls to a functions
# if they use direct funcargs (i.e. direct parametrization) # if they use direct funcargs (i.e. direct parametrization)
# because we want later test execution to be able to rely on # because we want later test execution to be able to rely on
@ -124,8 +153,8 @@ def add_funcarg_pseudo_fixture_def(collector, metafunc, fixturemanager):
if not metafunc._calls[0].funcargs: if not metafunc._calls[0].funcargs:
return # this function call does not have direct parametrization return # this function call does not have direct parametrization
# collect funcargs of all callspecs into a list of values # collect funcargs of all callspecs into a list of values
arg2params = {} arg2params = {} # type: Dict[str, List[object]]
arg2scope = {} arg2scope = {} # type: Dict[str, _Scope]
for callspec in metafunc._calls: for callspec in metafunc._calls:
for argname, argvalue in callspec.funcargs.items(): for argname, argvalue in callspec.funcargs.items():
assert argname not in callspec.params assert argname not in callspec.params
@ -233,7 +262,7 @@ def reorder_items(items):
return list(reorder_items_atscope(items, argkeys_cache, items_by_argkey, 0)) return list(reorder_items_atscope(items, argkeys_cache, items_by_argkey, 0))
def fix_cache_order(item, argkeys_cache, items_by_argkey): def fix_cache_order(item, argkeys_cache, items_by_argkey) -> None:
for scopenum in range(0, scopenum_function): for scopenum in range(0, scopenum_function):
for key in argkeys_cache[scopenum].get(item, []): for key in argkeys_cache[scopenum].get(item, []):
items_by_argkey[scopenum][key].appendleft(item) items_by_argkey[scopenum][key].appendleft(item)
@ -279,7 +308,7 @@ def reorder_items_atscope(items, argkeys_cache, items_by_argkey, scopenum):
return items_done return items_done
def fillfixtures(function): def fillfixtures(function) -> None:
""" fill missing funcargs for a test function. """ """ fill missing funcargs for a test function. """
warnings.warn(FILLFUNCARGS, stacklevel=2) warnings.warn(FILLFUNCARGS, stacklevel=2)
try: try:
@ -309,15 +338,15 @@ def get_direct_param_fixture_func(request):
@attr.s(slots=True) @attr.s(slots=True)
class FuncFixtureInfo: class FuncFixtureInfo:
# original function argument names # original function argument names
argnames = attr.ib(type=tuple) argnames = attr.ib(type=Tuple[str, ...])
# argnames that function immediately requires. These include argnames + # argnames that function immediately requires. These include argnames +
# fixture names specified via usefixtures and via autouse=True in fixture # fixture names specified via usefixtures and via autouse=True in fixture
# definitions. # definitions.
initialnames = attr.ib(type=tuple) initialnames = attr.ib(type=Tuple[str, ...])
names_closure = attr.ib() # List[str] names_closure = attr.ib(type=List[str])
name2fixturedefs = attr.ib() # List[str, List[FixtureDef]] name2fixturedefs = attr.ib(type=Dict[str, Sequence["FixtureDef"]])
def prune_dependency_tree(self): def prune_dependency_tree(self) -> None:
"""Recompute names_closure from initialnames and name2fixturedefs """Recompute names_closure from initialnames and name2fixturedefs
Can only reduce names_closure, which means that the new closure will Can only reduce names_closure, which means that the new closure will
@ -328,7 +357,7 @@ class FuncFixtureInfo:
tree. In this way the dependency tree can get pruned, and the closure tree. In this way the dependency tree can get pruned, and the closure
of argnames may get reduced. of argnames may get reduced.
""" """
closure = set() closure = set() # type: Set[str]
working_set = set(self.initialnames) working_set = set(self.initialnames)
while working_set: while working_set:
argname = working_set.pop() argname = working_set.pop()
@ -353,27 +382,29 @@ class FixtureRequest:
the fixture is parametrized indirectly. the fixture is parametrized indirectly.
""" """
def __init__(self, pyfuncitem): def __init__(self, pyfuncitem) -> None:
self._pyfuncitem = pyfuncitem self._pyfuncitem = pyfuncitem
#: fixture for which this request is being performed #: fixture for which this request is being performed
self.fixturename = None self.fixturename = None # type: Optional[str]
#: Scope string, one of "function", "class", "module", "session" #: Scope string, one of "function", "class", "module", "session"
self.scope = "function" self.scope = "function" # type: _Scope
self._fixture_defs = {} # type: Dict[str, FixtureDef] self._fixture_defs = {} # type: Dict[str, FixtureDef]
fixtureinfo = pyfuncitem._fixtureinfo fixtureinfo = pyfuncitem._fixtureinfo # type: FuncFixtureInfo
self._arg2fixturedefs = fixtureinfo.name2fixturedefs.copy() self._arg2fixturedefs = fixtureinfo.name2fixturedefs.copy()
self._arg2index = {} self._arg2index = {} # type: Dict[str, int]
self._fixturemanager = pyfuncitem.session._fixturemanager self._fixturemanager = (
pyfuncitem.session._fixturemanager
) # type: FixtureManager
@property @property
def fixturenames(self): def fixturenames(self) -> List[str]:
"""names of all active fixtures in this request""" """names of all active fixtures in this request"""
result = list(self._pyfuncitem._fixtureinfo.names_closure) result = list(self._pyfuncitem._fixtureinfo.names_closure)
result.extend(set(self._fixture_defs).difference(result)) result.extend(set(self._fixture_defs).difference(result))
return result return result
@property @property
def funcargnames(self): def funcargnames(self) -> List[str]:
""" alias attribute for ``fixturenames`` for pre-2.3 compatibility""" """ alias attribute for ``fixturenames`` for pre-2.3 compatibility"""
warnings.warn(FUNCARGNAMES, stacklevel=2) warnings.warn(FUNCARGNAMES, stacklevel=2)
return self.fixturenames return self.fixturenames
@ -383,15 +414,18 @@ class FixtureRequest:
""" underlying collection node (depends on current request scope)""" """ underlying collection node (depends on current request scope)"""
return self._getscopeitem(self.scope) return self._getscopeitem(self.scope)
def _getnextfixturedef(self, argname): def _getnextfixturedef(self, argname: str) -> "FixtureDef":
fixturedefs = self._arg2fixturedefs.get(argname, None) fixturedefs = self._arg2fixturedefs.get(argname, None)
if fixturedefs is None: if fixturedefs is None:
# we arrive here because of a dynamic call to # we arrive here because of a dynamic call to
# getfixturevalue(argname) usage which was naturally # getfixturevalue(argname) usage which was naturally
# not known at parsing/collection time # not known at parsing/collection time
assert self._pyfuncitem.parent is not None
parentid = self._pyfuncitem.parent.nodeid parentid = self._pyfuncitem.parent.nodeid
fixturedefs = self._fixturemanager.getfixturedefs(argname, parentid) fixturedefs = self._fixturemanager.getfixturedefs(argname, parentid)
self._arg2fixturedefs[argname] = fixturedefs # TODO: Fix this type ignore. Either add assert or adjust types.
# Can this be None here?
self._arg2fixturedefs[argname] = fixturedefs # type: ignore[assignment] # noqa: F821
# fixturedefs list is immutable so we maintain a decreasing index # fixturedefs list is immutable so we maintain a decreasing index
index = self._arg2index.get(argname, 0) - 1 index = self._arg2index.get(argname, 0) - 1
if fixturedefs is None or (-index > len(fixturedefs)): if fixturedefs is None or (-index > len(fixturedefs)):
@ -447,20 +481,20 @@ class FixtureRequest:
""" pytest session object. """ """ pytest session object. """
return self._pyfuncitem.session return self._pyfuncitem.session
def addfinalizer(self, finalizer): def addfinalizer(self, finalizer: Callable[[], object]) -> None:
""" add finalizer/teardown function to be called after the """ add finalizer/teardown function to be called after the
last test within the requesting test context finished last test within the requesting test context finished
execution. """ execution. """
# XXX usually this method is shadowed by fixturedef specific ones # XXX usually this method is shadowed by fixturedef specific ones
self._addfinalizer(finalizer, scope=self.scope) self._addfinalizer(finalizer, scope=self.scope)
def _addfinalizer(self, finalizer, scope): def _addfinalizer(self, finalizer: Callable[[], object], scope) -> None:
colitem = self._getscopeitem(scope) colitem = self._getscopeitem(scope)
self._pyfuncitem.session._setupstate.addfinalizer( self._pyfuncitem.session._setupstate.addfinalizer(
finalizer=finalizer, colitem=colitem finalizer=finalizer, colitem=colitem
) )
def applymarker(self, marker): def applymarker(self, marker) -> None:
""" Apply a marker to a single test function invocation. """ Apply a marker to a single test function invocation.
This method is useful if you don't want to have a keyword/marker This method is useful if you don't want to have a keyword/marker
on all function invocations. on all function invocations.
@ -470,18 +504,18 @@ class FixtureRequest:
""" """
self.node.add_marker(marker) self.node.add_marker(marker)
def raiseerror(self, msg): def raiseerror(self, msg: Optional[str]) -> "NoReturn":
""" raise a FixtureLookupError with the given message. """ """ raise a FixtureLookupError with the given message. """
raise self._fixturemanager.FixtureLookupError(None, self, msg) raise self._fixturemanager.FixtureLookupError(None, self, msg)
def _fillfixtures(self): def _fillfixtures(self) -> None:
item = self._pyfuncitem item = self._pyfuncitem
fixturenames = getattr(item, "fixturenames", self.fixturenames) fixturenames = getattr(item, "fixturenames", self.fixturenames)
for argname in fixturenames: for argname in fixturenames:
if argname not in item.funcargs: if argname not in item.funcargs:
item.funcargs[argname] = self.getfixturevalue(argname) item.funcargs[argname] = self.getfixturevalue(argname)
def getfixturevalue(self, argname): def getfixturevalue(self, argname: str) -> Any:
""" Dynamically run a named fixture function. """ Dynamically run a named fixture function.
Declaring fixtures via function argument is recommended where possible. Declaring fixtures via function argument is recommended where possible.
@ -492,9 +526,13 @@ class FixtureRequest:
:raise pytest.FixtureLookupError: :raise pytest.FixtureLookupError:
If the given fixture could not be found. If the given fixture could not be found.
""" """
return self._get_active_fixturedef(argname).cached_result[0] fixturedef = self._get_active_fixturedef(argname)
assert fixturedef.cached_result is not None
return fixturedef.cached_result[0]
def _get_active_fixturedef(self, argname): def _get_active_fixturedef(
self, argname: str
) -> Union["FixtureDef", PseudoFixtureDef]:
try: try:
return self._fixture_defs[argname] return self._fixture_defs[argname]
except KeyError: except KeyError:
@ -503,7 +541,7 @@ class FixtureRequest:
except FixtureLookupError: except FixtureLookupError:
if argname == "request": if argname == "request":
cached_result = (self, [0], None) cached_result = (self, [0], None)
scope = "function" scope = "function" # type: _Scope
return PseudoFixtureDef(cached_result, scope) return PseudoFixtureDef(cached_result, scope)
raise raise
# remove indent to prevent the python3 exception # remove indent to prevent the python3 exception
@ -512,15 +550,16 @@ class FixtureRequest:
self._fixture_defs[argname] = fixturedef self._fixture_defs[argname] = fixturedef
return fixturedef return fixturedef
def _get_fixturestack(self): def _get_fixturestack(self) -> List["FixtureDef"]:
current = self current = self
values = [] values = [] # type: List[FixtureDef]
while 1: while 1:
fixturedef = getattr(current, "_fixturedef", None) fixturedef = getattr(current, "_fixturedef", None)
if fixturedef is None: if fixturedef is None:
values.reverse() values.reverse()
return values return values
values.append(fixturedef) values.append(fixturedef)
assert isinstance(current, SubRequest)
current = current._parent_request current = current._parent_request
def _compute_fixture_value(self, fixturedef: "FixtureDef") -> None: def _compute_fixture_value(self, fixturedef: "FixtureDef") -> None:
@ -593,13 +632,15 @@ class FixtureRequest:
finally: finally:
self._schedule_finalizers(fixturedef, subrequest) self._schedule_finalizers(fixturedef, subrequest)
def _schedule_finalizers(self, fixturedef, subrequest): def _schedule_finalizers(
self, fixturedef: "FixtureDef", subrequest: "SubRequest"
) -> None:
# if fixture function failed it might have registered finalizers # if fixture function failed it might have registered finalizers
self.session._setupstate.addfinalizer( self.session._setupstate.addfinalizer(
functools.partial(fixturedef.finish, request=subrequest), subrequest.node functools.partial(fixturedef.finish, request=subrequest), subrequest.node
) )
def _check_scope(self, argname, invoking_scope, requested_scope): def _check_scope(self, argname, invoking_scope: "_Scope", requested_scope) -> None:
if argname == "request": if argname == "request":
return return
if scopemismatch(invoking_scope, requested_scope): if scopemismatch(invoking_scope, requested_scope):
@ -613,7 +654,7 @@ class FixtureRequest:
pytrace=False, pytrace=False,
) )
def _factorytraceback(self): def _factorytraceback(self) -> List[str]:
lines = [] lines = []
for fixturedef in self._get_fixturestack(): for fixturedef in self._get_fixturestack():
factory = fixturedef.func factory = fixturedef.func
@ -639,7 +680,7 @@ class FixtureRequest:
) )
return node return node
def __repr__(self): def __repr__(self) -> str:
return "<FixtureRequest for %r>" % (self.node) return "<FixtureRequest for %r>" % (self.node)
@ -647,9 +688,16 @@ class SubRequest(FixtureRequest):
""" a sub request for handling getting a fixture from a """ a sub request for handling getting a fixture from a
test function/fixture. """ test function/fixture. """
def __init__(self, request, scope, param, param_index, fixturedef): def __init__(
self,
request: "FixtureRequest",
scope: "_Scope",
param,
param_index: int,
fixturedef: "FixtureDef",
) -> None:
self._parent_request = request self._parent_request = request
self.fixturename = fixturedef.argname self.fixturename = fixturedef.argname # type: str
if param is not NOTSET: if param is not NOTSET:
self.param = param self.param = param
self.param_index = param_index self.param_index = param_index
@ -661,13 +709,15 @@ class SubRequest(FixtureRequest):
self._arg2index = request._arg2index self._arg2index = request._arg2index
self._fixturemanager = request._fixturemanager self._fixturemanager = request._fixturemanager
def __repr__(self): def __repr__(self) -> str:
return "<SubRequest {!r} for {!r}>".format(self.fixturename, self._pyfuncitem) return "<SubRequest {!r} for {!r}>".format(self.fixturename, self._pyfuncitem)
def addfinalizer(self, finalizer): def addfinalizer(self, finalizer: Callable[[], object]) -> None:
self._fixturedef.addfinalizer(finalizer) self._fixturedef.addfinalizer(finalizer)
def _schedule_finalizers(self, fixturedef, subrequest): def _schedule_finalizers(
self, fixturedef: "FixtureDef", subrequest: "SubRequest"
) -> None:
# if the executing fixturedef was not explicitly requested in the argument list (via # if the executing fixturedef was not explicitly requested in the argument list (via
# getfixturevalue inside the fixture call) then ensure this fixture def will be finished # getfixturevalue inside the fixture call) then ensure this fixture def will be finished
# first # first
@ -678,20 +728,21 @@ class SubRequest(FixtureRequest):
super()._schedule_finalizers(fixturedef, subrequest) super()._schedule_finalizers(fixturedef, subrequest)
scopes = "session package module class function".split() scopes = ["session", "package", "module", "class", "function"] # type: List[_Scope]
scopenum_function = scopes.index("function") scopenum_function = scopes.index("function")
def scopemismatch(currentscope, newscope): def scopemismatch(currentscope: "_Scope", newscope: "_Scope") -> bool:
return scopes.index(newscope) > scopes.index(currentscope) return scopes.index(newscope) > scopes.index(currentscope)
def scope2index(scope, descr, where=None): def scope2index(scope: str, descr: str, where: Optional[str] = None) -> int:
"""Look up the index of ``scope`` and raise a descriptive value error """Look up the index of ``scope`` and raise a descriptive value error
if not defined. if not defined.
""" """
strscopes = scopes # type: Sequence[str]
try: try:
return scopes.index(scope) return strscopes.index(scope)
except ValueError: except ValueError:
fail( fail(
"{} {}got an unexpected scope value '{}'".format( "{} {}got an unexpected scope value '{}'".format(
@ -704,7 +755,7 @@ def scope2index(scope, descr, where=None):
class FixtureLookupError(LookupError): class FixtureLookupError(LookupError):
""" could not return a requested Fixture (missing or invalid). """ """ could not return a requested Fixture (missing or invalid). """
def __init__(self, argname, request, msg=None): def __init__(self, argname, request, msg: Optional[str] = None) -> None:
self.argname = argname self.argname = argname
self.request = request self.request = request
self.fixturestack = request._get_fixturestack() self.fixturestack = request._get_fixturestack()
@ -782,14 +833,14 @@ class FixtureLookupErrorRepr(TerminalRepr):
tw.line("%s:%d" % (self.filename, self.firstlineno + 1)) tw.line("%s:%d" % (self.filename, self.firstlineno + 1))
def fail_fixturefunc(fixturefunc, msg): def fail_fixturefunc(fixturefunc, msg: str) -> "NoReturn":
fs, lineno = getfslineno(fixturefunc) fs, lineno = getfslineno(fixturefunc)
location = "{}:{}".format(fs, lineno + 1) location = "{}:{}".format(fs, lineno + 1)
source = _pytest._code.Source(fixturefunc) source = _pytest._code.Source(fixturefunc)
fail(msg + ":\n\n" + str(source.indent()) + "\n" + location, pytrace=False) fail(msg + ":\n\n" + str(source.indent()) + "\n" + location, pytrace=False)
def call_fixture_func(fixturefunc, request, kwargs): def call_fixture_func(fixturefunc, request: FixtureRequest, kwargs) -> object:
yieldctx = is_generator(fixturefunc) yieldctx = is_generator(fixturefunc)
if yieldctx: if yieldctx:
generator = fixturefunc(**kwargs) generator = fixturefunc(**kwargs)
@ -806,7 +857,7 @@ def call_fixture_func(fixturefunc, request, kwargs):
return fixture_result return fixture_result
def _teardown_yield_fixture(fixturefunc, it): def _teardown_yield_fixture(fixturefunc, it) -> None:
"""Executes the teardown of a fixture function by advancing the iterator after the """Executes the teardown of a fixture function by advancing the iterator after the
yield and ensure the iteration ends (if not it means there is more than one yield in the function)""" yield and ensure the iteration ends (if not it means there is more than one yield in the function)"""
try: try:
@ -819,7 +870,7 @@ def _teardown_yield_fixture(fixturefunc, it):
) )
def _eval_scope_callable(scope_callable, fixture_name, config): def _eval_scope_callable(scope_callable, fixture_name: str, config: Config) -> str:
try: try:
result = scope_callable(fixture_name=fixture_name, config=config) result = scope_callable(fixture_name=fixture_name, config=config)
except Exception: except Exception:
@ -843,15 +894,15 @@ class FixtureDef:
def __init__( def __init__(
self, self,
fixturemanager, fixturemanager: "FixtureManager",
baseid, baseid,
argname, argname: str,
func, func,
scope, scope: str,
params, params: Optional[Sequence[object]],
unittest=False, unittest: bool = False,
ids=None, ids=None,
): ) -> None:
self._fixturemanager = fixturemanager self._fixturemanager = fixturemanager
self.baseid = baseid or "" self.baseid = baseid or ""
self.has_location = baseid is not None self.has_location = baseid is not None
@ -859,23 +910,28 @@ class FixtureDef:
self.argname = argname self.argname = argname
if callable(scope): if callable(scope):
scope = _eval_scope_callable(scope, argname, fixturemanager.config) scope = _eval_scope_callable(scope, argname, fixturemanager.config)
self.scope = scope
self.scopenum = scope2index( self.scopenum = scope2index(
scope or "function", scope or "function",
descr="Fixture '{}'".format(func.__name__), descr="Fixture '{}'".format(func.__name__),
where=baseid, where=baseid,
) )
self.params = params # The cast is verified by scope2index.
self.argnames = getfuncargnames(func, name=argname, is_method=unittest) # (Some of the type annotations below are supposed to be inferred,
# but mypy 0.761 has some trouble without them.)
self.scope = cast("_Scope", scope) # type: _Scope
self.params = params # type: Optional[Sequence[object]]
self.argnames = getfuncargnames(
func, name=argname, is_method=unittest
) # type: Tuple[str, ...]
self.unittest = unittest self.unittest = unittest
self.ids = ids self.ids = ids
self.cached_result = None self.cached_result = None # type: Optional[_FixtureCachedResult]
self._finalizers = [] self._finalizers = [] # type: List[Callable[[], object]]
def addfinalizer(self, finalizer): def addfinalizer(self, finalizer: Callable[[], object]) -> None:
self._finalizers.append(finalizer) self._finalizers.append(finalizer)
def finish(self, request): def finish(self, request: SubRequest) -> None:
exc = None exc = None
try: try:
while self._finalizers: while self._finalizers:
@ -899,12 +955,14 @@ class FixtureDef:
self.cached_result = None self.cached_result = None
self._finalizers = [] self._finalizers = []
def execute(self, request): def execute(self, request: SubRequest):
# get required arguments and register our own finish() # get required arguments and register our own finish()
# with their finalization # with their finalization
for argname in self.argnames: for argname in self.argnames:
fixturedef = request._get_active_fixturedef(argname) fixturedef = request._get_active_fixturedef(argname)
if argname != "request": if argname != "request":
# PseudoFixtureDef is only for "request".
assert isinstance(fixturedef, FixtureDef)
fixturedef.addfinalizer(functools.partial(self.finish, request=request)) fixturedef.addfinalizer(functools.partial(self.finish, request=request))
my_cache_key = self.cache_key(request) my_cache_key = self.cache_key(request)
@ -926,16 +984,16 @@ class FixtureDef:
hook = self._fixturemanager.session.gethookproxy(request.node.fspath) hook = self._fixturemanager.session.gethookproxy(request.node.fspath)
return hook.pytest_fixture_setup(fixturedef=self, request=request) return hook.pytest_fixture_setup(fixturedef=self, request=request)
def cache_key(self, request): def cache_key(self, request: SubRequest) -> object:
return request.param_index if not hasattr(request, "param") else request.param return request.param_index if not hasattr(request, "param") else request.param
def __repr__(self): def __repr__(self) -> str:
return "<FixtureDef argname={!r} scope={!r} baseid={!r}>".format( return "<FixtureDef argname={!r} scope={!r} baseid={!r}>".format(
self.argname, self.scope, self.baseid self.argname, self.scope, self.baseid
) )
def resolve_fixture_function(fixturedef, request): def resolve_fixture_function(fixturedef: FixtureDef, request: FixtureRequest):
"""Gets the actual callable that can be called to obtain the fixture value, dealing with unittest-specific """Gets the actual callable that can be called to obtain the fixture value, dealing with unittest-specific
instances and bound methods. instances and bound methods.
""" """
@ -961,7 +1019,7 @@ def resolve_fixture_function(fixturedef, request):
return fixturefunc return fixturefunc
def pytest_fixture_setup(fixturedef, request): def pytest_fixture_setup(fixturedef: FixtureDef, request: SubRequest) -> object:
""" Execution of fixture setup. """ """ Execution of fixture setup. """
kwargs = {} kwargs = {}
for argname in fixturedef.argnames: for argname in fixturedef.argnames:
@ -976,7 +1034,9 @@ def pytest_fixture_setup(fixturedef, request):
try: try:
result = call_fixture_func(fixturefunc, request, kwargs) result = call_fixture_func(fixturefunc, request, kwargs)
except TEST_OUTCOME: except TEST_OUTCOME:
fixturedef.cached_result = (None, my_cache_key, sys.exc_info()) exc_info = sys.exc_info()
assert exc_info[0] is not None
fixturedef.cached_result = (None, my_cache_key, exc_info)
raise raise
fixturedef.cached_result = (result, my_cache_key, None) fixturedef.cached_result = (result, my_cache_key, None)
return result return result
@ -1190,7 +1250,7 @@ def yield_fixture(
@fixture(scope="session") @fixture(scope="session")
def pytestconfig(request): def pytestconfig(request: FixtureRequest):
"""Session-scoped fixture that returns the :class:`_pytest.config.Config` object. """Session-scoped fixture that returns the :class:`_pytest.config.Config` object.
Example:: Example::
@ -1247,15 +1307,17 @@ class FixtureManager:
FixtureLookupError = FixtureLookupError FixtureLookupError = FixtureLookupError
FixtureLookupErrorRepr = FixtureLookupErrorRepr FixtureLookupErrorRepr = FixtureLookupErrorRepr
def __init__(self, session): def __init__(self, session: "Session") -> None:
self.session = session self.session = session
self.config = session.config self.config = session.config # type: Config
self._arg2fixturedefs = {} self._arg2fixturedefs = {} # type: Dict[str, List[FixtureDef]]
self._holderobjseen = set() self._holderobjseen = set() # type: Set
self._nodeid_and_autousenames = [("", self.config.getini("usefixtures"))] self._nodeid_and_autousenames = [
("", self.config.getini("usefixtures"))
] # type: List[Tuple[str, List[str]]]
session.config.pluginmanager.register(self, "funcmanage") session.config.pluginmanager.register(self, "funcmanage")
def _get_direct_parametrize_args(self, node): def _get_direct_parametrize_args(self, node) -> List[str]:
"""This function returns all the direct parametrization """This function returns all the direct parametrization
arguments of a node, so we don't mistake them for fixtures arguments of a node, so we don't mistake them for fixtures
@ -1264,7 +1326,7 @@ class FixtureManager:
This things are done later as well when dealing with parametrization This things are done later as well when dealing with parametrization
so this could be improved so this could be improved
""" """
parametrize_argnames = [] parametrize_argnames = [] # type: List[str]
for marker in node.iter_markers(name="parametrize"): for marker in node.iter_markers(name="parametrize"):
if not marker.kwargs.get("indirect", False): if not marker.kwargs.get("indirect", False):
p_argnames, _ = ParameterSet._parse_parametrize_args( p_argnames, _ = ParameterSet._parse_parametrize_args(
@ -1274,7 +1336,7 @@ class FixtureManager:
return parametrize_argnames return parametrize_argnames
def getfixtureinfo(self, node, func, cls, funcargs=True): def getfixtureinfo(self, node, func, cls, funcargs: bool = True) -> FuncFixtureInfo:
if funcargs and not getattr(node, "nofuncargs", False): if funcargs and not getattr(node, "nofuncargs", False):
argnames = getfuncargnames(func, name=node.name, cls=cls) argnames = getfuncargnames(func, name=node.name, cls=cls)
else: else:
@ -1290,10 +1352,10 @@ class FixtureManager:
) )
return FuncFixtureInfo(argnames, initialnames, names_closure, arg2fixturedefs) return FuncFixtureInfo(argnames, initialnames, names_closure, arg2fixturedefs)
def pytest_plugin_registered(self, plugin): def pytest_plugin_registered(self, plugin: _PluggyPlugin) -> None:
nodeid = None nodeid = None
try: try:
p = py.path.local(plugin.__file__).realpath() p = py.path.local(plugin.__file__).realpath() # type: ignore[attr-defined] # noqa: F821
except AttributeError: except AttributeError:
pass pass
else: else:
@ -1309,9 +1371,9 @@ class FixtureManager:
self.parsefactories(plugin, nodeid) self.parsefactories(plugin, nodeid)
def _getautousenames(self, nodeid): def _getautousenames(self, nodeid: str) -> List[str]:
""" return a tuple of fixture names to be used. """ """ return a tuple of fixture names to be used. """
autousenames = [] autousenames = [] # type: List[str]
for baseid, basenames in self._nodeid_and_autousenames: for baseid, basenames in self._nodeid_and_autousenames:
if nodeid.startswith(baseid): if nodeid.startswith(baseid):
if baseid: if baseid:
@ -1322,7 +1384,9 @@ class FixtureManager:
autousenames.extend(basenames) autousenames.extend(basenames)
return autousenames return autousenames
def getfixtureclosure(self, fixturenames, parentnode, ignore_args=()): def getfixtureclosure(
self, fixturenames: Tuple[str, ...], parentnode, ignore_args: Sequence[str] = ()
) -> Tuple[Tuple[str, ...], List[str], Dict[str, Sequence[FixtureDef]]]:
# collect the closure of all fixtures , starting with the given # collect the closure of all fixtures , starting with the given
# fixturenames as the initial set. As we have to visit all # fixturenames as the initial set. As we have to visit all
# factory definitions anyway, we also return an arg2fixturedefs # factory definitions anyway, we also return an arg2fixturedefs
@ -1333,7 +1397,7 @@ class FixtureManager:
parentid = parentnode.nodeid parentid = parentnode.nodeid
fixturenames_closure = self._getautousenames(parentid) fixturenames_closure = self._getautousenames(parentid)
def merge(otherlist): def merge(otherlist: Iterable[str]) -> None:
for arg in otherlist: for arg in otherlist:
if arg not in fixturenames_closure: if arg not in fixturenames_closure:
fixturenames_closure.append(arg) fixturenames_closure.append(arg)
@ -1345,7 +1409,7 @@ class FixtureManager:
# need to return it as well, so save this. # need to return it as well, so save this.
initialnames = tuple(fixturenames_closure) initialnames = tuple(fixturenames_closure)
arg2fixturedefs = {} arg2fixturedefs = {} # type: Dict[str, Sequence[FixtureDef]]
lastlen = -1 lastlen = -1
while lastlen != len(fixturenames_closure): while lastlen != len(fixturenames_closure):
lastlen = len(fixturenames_closure) lastlen = len(fixturenames_closure)
@ -1359,7 +1423,7 @@ class FixtureManager:
arg2fixturedefs[argname] = fixturedefs arg2fixturedefs[argname] = fixturedefs
merge(fixturedefs[-1].argnames) merge(fixturedefs[-1].argnames)
def sort_by_scope(arg_name): def sort_by_scope(arg_name: str) -> int:
try: try:
fixturedefs = arg2fixturedefs[arg_name] fixturedefs = arg2fixturedefs[arg_name]
except KeyError: except KeyError:
@ -1370,7 +1434,7 @@ class FixtureManager:
fixturenames_closure.sort(key=sort_by_scope) fixturenames_closure.sort(key=sort_by_scope)
return initialnames, fixturenames_closure, arg2fixturedefs return initialnames, fixturenames_closure, arg2fixturedefs
def pytest_generate_tests(self, metafunc): def pytest_generate_tests(self, metafunc: "Metafunc") -> None:
for argname in metafunc.fixturenames: for argname in metafunc.fixturenames:
faclist = metafunc._arg2fixturedefs.get(argname) faclist = metafunc._arg2fixturedefs.get(argname)
if faclist: if faclist:
@ -1404,7 +1468,9 @@ class FixtureManager:
# separate parametrized setups # separate parametrized setups
items[:] = reorder_items(items) items[:] = reorder_items(items)
def parsefactories(self, node_or_obj, nodeid=NOTSET, unittest=False): def parsefactories(
self, node_or_obj, nodeid=NOTSET, unittest: bool = False
) -> None:
if nodeid is not NOTSET: if nodeid is not NOTSET:
holderobj = node_or_obj holderobj = node_or_obj
else: else:
@ -1460,7 +1526,9 @@ class FixtureManager:
if autousenames: if autousenames:
self._nodeid_and_autousenames.append((nodeid or "", autousenames)) self._nodeid_and_autousenames.append((nodeid or "", autousenames))
def getfixturedefs(self, argname, nodeid): def getfixturedefs(
self, argname: str, nodeid: str
) -> Optional[Sequence[FixtureDef]]:
""" """
Gets a list of fixtures which are applicable to the given node id. Gets a list of fixtures which are applicable to the given node id.
@ -1474,7 +1542,9 @@ class FixtureManager:
return None return None
return tuple(self._matchfactories(fixturedefs, nodeid)) return tuple(self._matchfactories(fixturedefs, nodeid))
def _matchfactories(self, fixturedefs, nodeid): def _matchfactories(
self, fixturedefs: Iterable[FixtureDef], nodeid: str
) -> Iterator[FixtureDef]:
from _pytest import nodes from _pytest import nodes
for fixturedef in fixturedefs: for fixturedef in fixturedefs:

View File

@ -20,6 +20,8 @@ if TYPE_CHECKING:
from _pytest.config import PytestPluginManager from _pytest.config import PytestPluginManager
from _pytest.config import _PluggyPlugin from _pytest.config import _PluggyPlugin
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureDef
from _pytest.fixtures import SubRequest
from _pytest.main import Session from _pytest.main import Session
from _pytest.nodes import Collector from _pytest.nodes import Collector
from _pytest.nodes import Item from _pytest.nodes import Item
@ -450,7 +452,9 @@ def pytest_report_from_serializable(config: "Config", data):
@hookspec(firstresult=True) @hookspec(firstresult=True)
def pytest_fixture_setup(fixturedef, request): def pytest_fixture_setup(
fixturedef: "FixtureDef", request: "SubRequest"
) -> Optional[object]:
""" performs fixture setup execution. """ performs fixture setup execution.
:return: The return value of the call to the fixture function :return: The return value of the call to the fixture function
@ -464,7 +468,9 @@ def pytest_fixture_setup(fixturedef, request):
""" """
def pytest_fixture_post_finalizer(fixturedef, request): def pytest_fixture_post_finalizer(
fixturedef: "FixtureDef", request: "SubRequest"
) -> None:
"""Called after fixture teardown, but before the cache is cleared, so """Called after fixture teardown, but before the cache is cleared, so
the fixture result ``fixturedef.cached_result`` is still available (not the fixture result ``fixturedef.cached_result`` is still available (not
``None``).""" ``None``)."""

View File

@ -65,6 +65,7 @@ from _pytest.warning_types import PytestUnhandledCoroutineWarning
if TYPE_CHECKING: if TYPE_CHECKING:
from typing_extensions import Literal from typing_extensions import Literal
from _pytest.fixtures import _Scope
def pytest_addoption(parser: Parser) -> None: def pytest_addoption(parser: Parser) -> None:
@ -905,7 +906,7 @@ class Metafunc:
Callable[[object], Optional[object]], Callable[[object], Optional[object]],
] ]
] = None, ] = None,
scope: "Optional[str]" = None, scope: "Optional[_Scope]" = None,
*, *,
_param_mark: Optional[Mark] = None _param_mark: Optional[Mark] = None
) -> None: ) -> None:

View File

@ -1,3 +1,4 @@
from typing import Generator
from typing import Optional from typing import Optional
from typing import Union from typing import Union
@ -6,6 +7,8 @@ from _pytest._io.saferepr import saferepr
from _pytest.config import Config from _pytest.config import Config
from _pytest.config import ExitCode from _pytest.config import ExitCode
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureDef
from _pytest.fixtures import SubRequest
def pytest_addoption(parser: Parser) -> None: def pytest_addoption(parser: Parser) -> None:
@ -25,7 +28,9 @@ def pytest_addoption(parser: Parser) -> None:
@pytest.hookimpl(hookwrapper=True) @pytest.hookimpl(hookwrapper=True)
def pytest_fixture_setup(fixturedef, request): def pytest_fixture_setup(
fixturedef: FixtureDef, request: SubRequest
) -> Generator[None, None, None]:
yield yield
if request.config.option.setupshow: if request.config.option.setupshow:
if hasattr(request, "param"): if hasattr(request, "param"):
@ -33,24 +38,25 @@ def pytest_fixture_setup(fixturedef, request):
# display it now and during the teardown (in .finish()). # display it now and during the teardown (in .finish()).
if fixturedef.ids: if fixturedef.ids:
if callable(fixturedef.ids): if callable(fixturedef.ids):
fixturedef.cached_param = fixturedef.ids(request.param) param = fixturedef.ids(request.param)
else: else:
fixturedef.cached_param = fixturedef.ids[request.param_index] param = fixturedef.ids[request.param_index]
else: else:
fixturedef.cached_param = request.param param = request.param
fixturedef.cached_param = param # type: ignore[attr-defined] # noqa: F821
_show_fixture_action(fixturedef, "SETUP") _show_fixture_action(fixturedef, "SETUP")
def pytest_fixture_post_finalizer(fixturedef) -> None: def pytest_fixture_post_finalizer(fixturedef: FixtureDef) -> None:
if fixturedef.cached_result is not None: if fixturedef.cached_result is not None:
config = fixturedef._fixturemanager.config config = fixturedef._fixturemanager.config
if config.option.setupshow: if config.option.setupshow:
_show_fixture_action(fixturedef, "TEARDOWN") _show_fixture_action(fixturedef, "TEARDOWN")
if hasattr(fixturedef, "cached_param"): if hasattr(fixturedef, "cached_param"):
del fixturedef.cached_param del fixturedef.cached_param # type: ignore[attr-defined] # noqa: F821
def _show_fixture_action(fixturedef, msg): def _show_fixture_action(fixturedef: FixtureDef, msg: str) -> None:
config = fixturedef._fixturemanager.config config = fixturedef._fixturemanager.config
capman = config.pluginmanager.getplugin("capturemanager") capman = config.pluginmanager.getplugin("capturemanager")
if capman: if capman:
@ -73,7 +79,7 @@ def _show_fixture_action(fixturedef, msg):
tw.write(" (fixtures used: {})".format(", ".join(deps))) tw.write(" (fixtures used: {})".format(", ".join(deps)))
if hasattr(fixturedef, "cached_param"): if hasattr(fixturedef, "cached_param"):
tw.write("[{}]".format(saferepr(fixturedef.cached_param, maxsize=42))) tw.write("[{}]".format(saferepr(fixturedef.cached_param, maxsize=42))) # type: ignore[attr-defined]
tw.flush() tw.flush()

View File

@ -5,6 +5,8 @@ import pytest
from _pytest.config import Config from _pytest.config import Config
from _pytest.config import ExitCode from _pytest.config import ExitCode
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureDef
from _pytest.fixtures import SubRequest
def pytest_addoption(parser: Parser) -> None: def pytest_addoption(parser: Parser) -> None:
@ -19,7 +21,9 @@ def pytest_addoption(parser: Parser) -> None:
@pytest.hookimpl(tryfirst=True) @pytest.hookimpl(tryfirst=True)
def pytest_fixture_setup(fixturedef, request): def pytest_fixture_setup(
fixturedef: FixtureDef, request: SubRequest
) -> Optional[object]:
# Will return a dummy fixture if the setuponly option is provided. # Will return a dummy fixture if the setuponly option is provided.
if request.config.option.setupplan: if request.config.option.setupplan:
my_cache_key = fixturedef.cache_key(request) my_cache_key = fixturedef.cache_key(request)

View File

@ -113,7 +113,7 @@ class TestMetafunc:
fail.Exception, fail.Exception,
match=r"parametrize\(\) call in func got an unexpected scope value 'doggy'", match=r"parametrize\(\) call in func got an unexpected scope value 'doggy'",
): ):
metafunc.parametrize("x", [1], scope="doggy") metafunc.parametrize("x", [1], scope="doggy") # type: ignore[arg-type] # noqa: F821
def test_parametrize_request_name(self, testdir: Testdir) -> None: def test_parametrize_request_name(self, testdir: Testdir) -> None:
"""Show proper error when 'request' is used as a parameter name in parametrize (#6183)""" """Show proper error when 'request' is used as a parameter name in parametrize (#6183)"""