Type annotate some more hooks & impls

This commit is contained in:
Ran Benita 2020-05-01 14:40:15 +03:00
parent ef34729541
commit 247c4c0482
23 changed files with 175 additions and 99 deletions

View File

@ -4,6 +4,7 @@ support for presenting detailed information in failing assertions.
import sys
from typing import Any
from typing import List
from typing import Generator
from typing import Optional
from _pytest.assertion import rewrite
@ -14,6 +15,7 @@ from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config import hookimpl
from _pytest.config.argparsing import Parser
from _pytest.nodes import Item
if TYPE_CHECKING:
from _pytest.main import Session
@ -113,7 +115,7 @@ def pytest_collection(session: "Session") -> None:
@hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_protocol(item):
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
"""Setup the pytest_assertrepr_compare and pytest_assertion_pass hooks
The rewrite module will use util._reprcompare if
@ -122,8 +124,7 @@ def pytest_runtest_protocol(item):
comparison for the test.
"""
def callbinrepr(op, left, right):
# type: (str, object, object) -> Optional[str]
def callbinrepr(op, left: object, right: object) -> Optional[str]:
"""Call the pytest_assertrepr_compare hook and prepare the result
This uses the first result from the hook and then ensures the
@ -156,7 +157,7 @@ def pytest_runtest_protocol(item):
if item.ihook.pytest_assertion_pass.get_hookimpls():
def call_assertion_pass_hook(lineno, orig, expl):
def call_assertion_pass_hook(lineno: int, orig: str, expl: str) -> None:
item.ihook.pytest_assertion_pass(
item=item, lineno=lineno, orig=orig, expl=expl
)

View File

@ -444,14 +444,12 @@ def _call_reprcompare(ops, results, expls, each_obj):
return expl
def _call_assertion_pass(lineno, orig, expl):
# type: (int, str, str) -> None
def _call_assertion_pass(lineno: int, orig: str, expl: str) -> None:
if util._assertion_pass is not None:
util._assertion_pass(lineno, orig, expl)
def _check_if_assertion_pass_impl():
# type: () -> bool
def _check_if_assertion_pass_impl() -> bool:
"""Checks if any plugins implement the pytest_assertion_pass hook
in order not to generate explanation unecessarily (might be expensive)"""
return True if util._assertion_pass else False

View File

@ -29,6 +29,7 @@ from _pytest.config import ExitCode
from _pytest.config.argparsing import Parser
from _pytest.main import Session
from _pytest.python import Module
from _pytest.reports import TestReport
README_CONTENT = """\
# pytest cache directory #
@ -265,7 +266,7 @@ class LFPlugin:
if self.active and self.config.getoption("verbose") >= 0:
return "run-last-failure: %s" % self._report_status
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: TestReport) -> None:
if (report.when == "call" and report.passed) or report.skipped:
self.lastfailed.pop(report.nodeid, None)
elif report.failed:

View File

@ -17,6 +17,9 @@ import pytest
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.fixtures import SubRequest
from _pytest.nodes import Collector
from _pytest.nodes import Item
if TYPE_CHECKING:
from typing_extensions import Literal
@ -710,7 +713,7 @@ class CaptureManager:
# Hooks
@pytest.hookimpl(hookwrapper=True)
def pytest_make_collect_report(self, collector):
def pytest_make_collect_report(self, collector: Collector):
if isinstance(collector, pytest.File):
self.resume_global_capture()
outcome = yield
@ -725,17 +728,17 @@ class CaptureManager:
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(self, item):
def pytest_runtest_setup(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("setup", item):
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(self, item):
def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("call", item):
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(self, item):
def pytest_runtest_teardown(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("teardown", item):
yield

View File

@ -4,12 +4,18 @@ import functools
import sys
from _pytest import outcomes
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config import ConftestImportFailure
from _pytest.config import hookimpl
from _pytest.config import PytestPluginManager
from _pytest.config.argparsing import Parser
from _pytest.config.exceptions import UsageError
from _pytest.nodes import Node
from _pytest.reports import BaseReport
if TYPE_CHECKING:
from _pytest.runner import CallInfo
def _validate_usepdb_cls(value):
@ -259,7 +265,9 @@ class pytestPDB:
class PdbInvoke:
def pytest_exception_interact(self, node, call, report):
def pytest_exception_interact(
self, node: Node, call: "CallInfo", report: BaseReport
) -> None:
capman = node.config.pluginmanager.getplugin("capturemanager")
if capman:
capman.suspend_global_capture(in_=True)
@ -306,7 +314,7 @@ def maybe_wrap_pytest_function_for_tracing(pyfuncitem):
wrap_pytest_function_for_tracing(pyfuncitem)
def _enter_pdb(node, excinfo, rep):
def _enter_pdb(node: Node, excinfo, rep: BaseReport) -> BaseReport:
# XXX we re-use the TerminalReporter's terminalwriter
# because this seems to avoid some encoding related troubles
# for not completely clear reasons.
@ -330,7 +338,7 @@ def _enter_pdb(node, excinfo, rep):
rep.toterminal(tw)
tw.sep(">", "entering PDB")
tb = _postmortem_traceback(excinfo)
rep._pdbshown = True
rep._pdbshown = True # type: ignore[attr-defined] # noqa: F821
post_mortem(tb)
return rep

View File

@ -1,11 +1,13 @@
import io
import os
import sys
from typing import Generator
from typing import TextIO
import pytest
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.nodes import Item
from _pytest.store import StoreKey
@ -82,7 +84,7 @@ class FaultHandlerHooks:
return float(config.getini("faulthandler_timeout") or 0.0)
@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_runtest_protocol(self, item):
def pytest_runtest_protocol(self, item: Item) -> Generator[None, None, None]:
timeout = self.get_timeout_config_value(item.config)
stderr = item.config._store[fault_handler_stderr_key]
if timeout > 0 and stderr is not None:
@ -105,7 +107,7 @@ class FaultHandlerHooks:
faulthandler.cancel_dump_traceback_later()
@pytest.hookimpl(tryfirst=True)
def pytest_exception_interact(self):
def pytest_exception_interact(self) -> None:
"""Cancel any traceback dumping due to an interactive exception being
raised.
"""

View File

@ -25,10 +25,16 @@ if TYPE_CHECKING:
from _pytest.main import Session
from _pytest.nodes import Collector
from _pytest.nodes import Item
from _pytest.nodes import Node
from _pytest.python import Function
from _pytest.python import Metafunc
from _pytest.python import Module
from _pytest.python import PyCollector
from _pytest.reports import BaseReport
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
from _pytest.runner import CallInfo
from _pytest.terminal import TerminalReporter
hookspec = HookspecMarker("pytest")
@ -268,7 +274,7 @@ def pytest_collect_file(path: py.path.local, parent) -> "Optional[Collector]":
# logging hooks for collection
def pytest_collectstart(collector):
def pytest_collectstart(collector: "Collector") -> None:
""" collector starts collecting. """
@ -285,7 +291,7 @@ def pytest_deselected(items):
@hookspec(firstresult=True)
def pytest_make_collect_report(collector):
def pytest_make_collect_report(collector: "Collector") -> "Optional[CollectReport]":
""" perform ``collector.collect()`` and return a CollectReport.
Stops at first non-None result, see :ref:`firstresult` """
@ -319,7 +325,7 @@ def pytest_pycollect_makeitem(
@hookspec(firstresult=True)
def pytest_pyfunc_call(pyfuncitem):
def pytest_pyfunc_call(pyfuncitem: "Function") -> Optional[object]:
""" call underlying test function.
Stops at first non-None result, see :ref:`firstresult` """
@ -330,7 +336,9 @@ def pytest_generate_tests(metafunc: "Metafunc") -> None:
@hookspec(firstresult=True)
def pytest_make_parametrize_id(config: "Config", val, argname) -> Optional[str]:
def pytest_make_parametrize_id(
config: "Config", val: object, argname: str
) -> Optional[str]:
"""Return a user-friendly string representation of the given ``val`` that will be used
by @pytest.mark.parametrize calls. Return None if the hook doesn't know about ``val``.
The parameter name is available as ``argname``, if required.
@ -349,7 +357,7 @@ def pytest_make_parametrize_id(config: "Config", val, argname) -> Optional[str]:
@hookspec(firstresult=True)
def pytest_runtestloop(session: "Session"):
def pytest_runtestloop(session: "Session") -> Optional[object]:
""" called for performing the main runtest loop
(after collection finished).
@ -360,7 +368,9 @@ def pytest_runtestloop(session: "Session"):
@hookspec(firstresult=True)
def pytest_runtest_protocol(item, nextitem):
def pytest_runtest_protocol(
item: "Item", nextitem: "Optional[Item]"
) -> Optional[object]:
""" implements the runtest_setup/call/teardown protocol for
the given test item, including capturing exceptions and calling
reporting hooks.
@ -399,15 +409,15 @@ def pytest_runtest_logfinish(nodeid, location):
"""
def pytest_runtest_setup(item):
def pytest_runtest_setup(item: "Item") -> None:
""" called before ``pytest_runtest_call(item)``. """
def pytest_runtest_call(item):
def pytest_runtest_call(item: "Item") -> None:
""" called to execute the test ``item``. """
def pytest_runtest_teardown(item, nextitem):
def pytest_runtest_teardown(item: "Item", nextitem: "Optional[Item]") -> None:
""" called after ``pytest_runtest_call``.
:arg nextitem: the scheduled-to-be-next test item (None if no further
@ -418,7 +428,7 @@ def pytest_runtest_teardown(item, nextitem):
@hookspec(firstresult=True)
def pytest_runtest_makereport(item, call):
def pytest_runtest_makereport(item: "Item", call: "CallInfo") -> Optional[object]:
""" return a :py:class:`_pytest.runner.TestReport` object
for the given :py:class:`pytest.Item <_pytest.main.Item>` and
:py:class:`_pytest.runner.CallInfo`.
@ -426,7 +436,7 @@ def pytest_runtest_makereport(item, call):
Stops at first non-None result, see :ref:`firstresult` """
def pytest_runtest_logreport(report):
def pytest_runtest_logreport(report: "TestReport") -> None:
""" process a test setup/call/teardown report relating to
the respective phase of executing a test. """
@ -511,7 +521,9 @@ def pytest_unconfigure(config: "Config") -> None:
# -------------------------------------------------------------------------
def pytest_assertrepr_compare(config: "Config", op, left, right):
def pytest_assertrepr_compare(
config: "Config", op: str, left: object, right: object
) -> Optional[List[str]]:
"""return explanation for comparisons in failing assert expressions.
Return None for no custom explanation, otherwise return a list
@ -523,7 +535,7 @@ def pytest_assertrepr_compare(config: "Config", op, left, right):
"""
def pytest_assertion_pass(item, lineno, orig, expl):
def pytest_assertion_pass(item, lineno: int, orig: str, expl: str) -> None:
"""
**(Experimental)**
@ -637,7 +649,9 @@ def pytest_report_teststatus(
"""
def pytest_terminal_summary(terminalreporter, exitstatus, config: "Config"):
def pytest_terminal_summary(
terminalreporter: "TerminalReporter", exitstatus: "ExitCode", config: "Config",
) -> None:
"""Add a section to terminal summary reporting.
:param _pytest.terminal.TerminalReporter terminalreporter: the internal terminal reporter object
@ -741,7 +755,9 @@ def pytest_keyboard_interrupt(excinfo):
""" called for keyboard interrupt. """
def pytest_exception_interact(node, call, report):
def pytest_exception_interact(
node: "Node", call: "CallInfo", report: "BaseReport"
) -> None:
"""called when an exception was raised which can potentially be
interactively handled.

View File

@ -24,7 +24,9 @@ from _pytest import timing
from _pytest.config import Config
from _pytest.config import filename_arg
from _pytest.config.argparsing import Parser
from _pytest.reports import TestReport
from _pytest.store import StoreKey
from _pytest.terminal import TerminalReporter
from _pytest.warnings import _issue_warning_captured
@ -517,7 +519,7 @@ class LogXML:
reporter.record_testreport(report)
return reporter
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: TestReport) -> None:
"""handle a setup/call/teardown report, generating the appropriate
xml tags as necessary.
@ -661,7 +663,7 @@ class LogXML:
logfile.write(Junit.testsuites([suite_node]).unicode(indent=0))
logfile.close()
def pytest_terminal_summary(self, terminalreporter):
def pytest_terminal_summary(self, terminalreporter: TerminalReporter) -> None:
terminalreporter.write_sep("-", "generated xml file: {}".format(self.logfile))
def add_global_property(self, name, value):

View File

@ -20,6 +20,7 @@ from _pytest.config import _strtobool
from _pytest.config import Config
from _pytest.config import create_terminal_writer
from _pytest.config.argparsing import Parser
from _pytest.main import Session
from _pytest.pathlib import Path
from _pytest.store import StoreKey
@ -618,7 +619,7 @@ class LoggingPlugin:
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtestloop(self, session):
def pytest_runtestloop(self, session: Session) -> Generator[None, None, None]:
"""Runs all collected test items."""
if session.config.option.collectonly:
@ -655,20 +656,21 @@ class LoggingPlugin:
item.add_report_section(when, "log", log)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(self, item):
def pytest_runtest_setup(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("setup")
item._store[catch_log_records_key] = {}
empty = {} # type: Dict[str, List[logging.LogRecord]]
item._store[catch_log_records_key] = empty
yield from self._runtest_for(item, "setup")
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(self, item):
def pytest_runtest_call(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("call")
yield from self._runtest_for(item, "call")
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(self, item):
def pytest_runtest_teardown(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("teardown")
yield from self._runtest_for(item, "teardown")
@ -676,7 +678,7 @@ class LoggingPlugin:
del item._store[catch_log_handler_key]
@pytest.hookimpl
def pytest_runtest_logfinish(self):
def pytest_runtest_logfinish(self) -> None:
self.log_cli_handler.set_when("finish")
@pytest.hookimpl(hookwrapper=True, tryfirst=True)

View File

@ -31,6 +31,7 @@ from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureManager
from _pytest.outcomes import exit
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
from _pytest.runner import collect_one_node
from _pytest.runner import SetupState
@ -441,7 +442,7 @@ class Session(nodes.FSCollector):
raise self.Interrupted(self.shouldstop)
@hookimpl(tryfirst=True)
def pytest_runtest_logreport(self, report) -> None:
def pytest_runtest_logreport(self, report: TestReport) -> None:
if report.failed and not hasattr(report, "wasxfail"):
self.testsfailed += 1
maxfail = self.config.getvalue("maxfail")

View File

@ -2,6 +2,7 @@
from _pytest import python
from _pytest import unittest
from _pytest.config import hookimpl
from _pytest.nodes import Item
@hookimpl(trylast=True)
@ -20,7 +21,7 @@ def teardown_nose(item):
call_optional(item.parent.obj, "teardown")
def is_potential_nosetest(item):
def is_potential_nosetest(item: Item) -> bool:
# extra check needed since we do not do nose style setup/teardown
# on direct unittest style classes
return isinstance(item, python.Function) and not isinstance(

View File

@ -2,11 +2,14 @@
import tempfile
from io import StringIO
from typing import IO
from typing import Union
import pytest
from _pytest.config import Config
from _pytest.config import create_terminal_writer
from _pytest.config.argparsing import Parser
from _pytest.store import StoreKey
from _pytest.terminal import TerminalReporter
pastebinfile_key = StoreKey[IO[bytes]]()
@ -63,11 +66,11 @@ def pytest_unconfigure(config: Config) -> None:
tr.write_line("pastebin session-log: %s\n" % pastebinurl)
def create_new_paste(contents):
def create_new_paste(contents: Union[str, bytes]) -> str:
"""
Creates a new paste using bpaste.net service.
:contents: paste contents as utf-8 encoded bytes
:contents: paste contents string
:returns: url to the pasted contents or error message
"""
import re
@ -79,7 +82,7 @@ def create_new_paste(contents):
try:
response = (
urlopen(url, data=urlencode(params).encode("ascii")).read().decode("utf-8")
)
) # type: str
except OSError as exc_info: # urllib errors
return "bad response: %s" % exc_info
m = re.search(r'href="/raw/(\w+)"', response)
@ -89,23 +92,20 @@ def create_new_paste(contents):
return "bad response: invalid format ('" + response + "')"
def pytest_terminal_summary(terminalreporter):
import _pytest.config
def pytest_terminal_summary(terminalreporter: TerminalReporter) -> None:
if terminalreporter.config.option.pastebin != "failed":
return
tr = terminalreporter
if "failed" in tr.stats:
if "failed" in terminalreporter.stats:
terminalreporter.write_sep("=", "Sending information to Paste Service")
for rep in terminalreporter.stats.get("failed"):
for rep in terminalreporter.stats["failed"]:
try:
msg = rep.longrepr.reprtraceback.reprentries[-1].reprfileloc
except AttributeError:
msg = tr._getfailureheadline(rep)
msg = terminalreporter._getfailureheadline(rep)
file = StringIO()
tw = _pytest.config.create_terminal_writer(terminalreporter.config, file)
tw = create_terminal_writer(terminalreporter.config, file)
rep.toterminal(tw)
s = file.getvalue()
assert len(s)
pastebinurl = create_new_paste(s)
tr.write_line("{} --> {}".format(msg, pastebinurl))
terminalreporter.write_line("{} --> {}".format(msg, pastebinurl))

View File

@ -12,6 +12,7 @@ from fnmatch import fnmatch
from io import StringIO
from typing import Callable
from typing import Dict
from typing import Generator
from typing import Iterable
from typing import List
from typing import Optional
@ -138,7 +139,7 @@ class LsofFdLeakChecker:
return True
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_protocol(self, item):
def pytest_runtest_protocol(self, item: Item) -> Generator[None, None, None]:
lines1 = self.get_open_files()
yield
if hasattr(sys, "pypy_version_info"):

View File

@ -173,7 +173,7 @@ def async_warn_and_skip(nodeid: str) -> None:
@hookimpl(trylast=True)
def pytest_pyfunc_call(pyfuncitem: "Function"):
def pytest_pyfunc_call(pyfuncitem: "Function") -> Optional[object]:
testfunction = pyfuncitem.obj
if is_async_function(testfunction):
async_warn_and_skip(pyfuncitem.nodeid)

View File

@ -26,6 +26,9 @@ from _pytest.nodes import Item
from _pytest.outcomes import skip
from _pytest.pathlib import Path
if TYPE_CHECKING:
from _pytest.runner import CallInfo
def getslaveinfoline(node):
try:
@ -42,7 +45,8 @@ def getslaveinfoline(node):
class BaseReport:
when = None # type: Optional[str]
location = None # type: Optional[Tuple[str, Optional[int], str]]
longrepr = None
# TODO: Improve this Any.
longrepr = None # type: Optional[Any]
sections = [] # type: List[Tuple[str, str]]
nodeid = None # type: str
@ -270,7 +274,7 @@ class TestReport(BaseReport):
)
@classmethod
def from_item_and_call(cls, item, call) -> "TestReport":
def from_item_and_call(cls, item: Item, call: "CallInfo") -> "TestReport":
"""
Factory method to create and fill a TestReport with standard item and call info.
"""
@ -281,7 +285,8 @@ class TestReport(BaseReport):
sections = []
if not call.excinfo:
outcome = "passed"
longrepr = None
# TODO: Improve this Any.
longrepr = None # type: Optional[Any]
else:
if not isinstance(excinfo, ExceptionInfo):
outcome = "failed"

View File

@ -7,6 +7,7 @@ import py
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.reports import TestReport
from _pytest.store import StoreKey
@ -66,7 +67,7 @@ class ResultLog:
testpath = report.fspath
self.write_log_entry(testpath, lettercode, longrepr)
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: TestReport) -> None:
if report.when != "call" and report.passed:
return
res = self.config.hook.pytest_report_teststatus(
@ -80,6 +81,7 @@ class ResultLog:
elif report.passed:
longrepr = ""
elif report.skipped:
assert report.longrepr is not None
longrepr = str(report.longrepr[2])
else:
longrepr = str(report.longrepr)

View File

@ -10,6 +10,7 @@ from typing import Tuple
import attr
from .reports import BaseReport
from .reports import CollectErrorRepr
from .reports import CollectReport
from .reports import TestReport
@ -19,6 +20,7 @@ from _pytest._code.code import ExceptionInfo
from _pytest.compat import TYPE_CHECKING
from _pytest.config.argparsing import Parser
from _pytest.nodes import Collector
from _pytest.nodes import Item
from _pytest.nodes import Node
from _pytest.outcomes import Exit
from _pytest.outcomes import Skipped
@ -29,6 +31,7 @@ if TYPE_CHECKING:
from typing_extensions import Literal
from _pytest.main import Session
from _pytest.terminal import TerminalReporter
#
# pytest plugin hooks
@ -46,7 +49,7 @@ def pytest_addoption(parser: Parser) -> None:
)
def pytest_terminal_summary(terminalreporter):
def pytest_terminal_summary(terminalreporter: "TerminalReporter") -> None:
durations = terminalreporter.config.option.durations
verbose = terminalreporter.config.getvalue("verbose")
if durations is None:
@ -86,17 +89,19 @@ def pytest_sessionfinish(session: "Session") -> None:
session._setupstate.teardown_all()
def pytest_runtest_protocol(item, nextitem):
def pytest_runtest_protocol(item: Item, nextitem: Optional[Item]) -> bool:
item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
runtestprotocol(item, nextitem=nextitem)
item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
return True
def runtestprotocol(item, log=True, nextitem=None):
def runtestprotocol(
item: Item, log: bool = True, nextitem: Optional[Item] = None
) -> List[TestReport]:
hasrequest = hasattr(item, "_request")
if hasrequest and not item._request:
item._initrequest()
if hasrequest and not item._request: # type: ignore[attr-defined] # noqa: F821
item._initrequest() # type: ignore[attr-defined] # noqa: F821
rep = call_and_report(item, "setup", log)
reports = [rep]
if rep.passed:
@ -108,12 +113,12 @@ def runtestprotocol(item, log=True, nextitem=None):
# after all teardown hooks have been called
# want funcargs and request info to go away
if hasrequest:
item._request = False
item.funcargs = None
item._request = False # type: ignore[attr-defined] # noqa: F821
item.funcargs = None # type: ignore[attr-defined] # noqa: F821
return reports
def show_test_item(item):
def show_test_item(item: Item) -> None:
"""Show test function, parameters and the fixtures of the test item."""
tw = item.config.get_terminal_writer()
tw.line()
@ -125,12 +130,12 @@ def show_test_item(item):
tw.flush()
def pytest_runtest_setup(item):
def pytest_runtest_setup(item: Item) -> None:
_update_current_test_var(item, "setup")
item.session._setupstate.prepare(item)
def pytest_runtest_call(item):
def pytest_runtest_call(item: Item) -> None:
_update_current_test_var(item, "call")
try:
del sys.last_type
@ -150,13 +155,15 @@ def pytest_runtest_call(item):
raise e
def pytest_runtest_teardown(item, nextitem):
def pytest_runtest_teardown(item: Item, nextitem: Optional[Item]) -> None:
_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):
def _update_current_test_var(
item: Item, when: Optional["Literal['setup', 'call', 'teardown']"]
) -> None:
"""
Update :envvar:`PYTEST_CURRENT_TEST` to reflect the current item and stage.
@ -188,11 +195,11 @@ def pytest_report_teststatus(report):
def call_and_report(
item, when: "Literal['setup', 'call', 'teardown']", log=True, **kwds
):
item: Item, when: "Literal['setup', 'call', 'teardown']", log: bool = True, **kwds
) -> TestReport:
call = call_runtest_hook(item, when, **kwds)
hook = item.ihook
report = hook.pytest_runtest_makereport(item=item, call=call)
report = hook.pytest_runtest_makereport(item=item, call=call) # type: TestReport
if log:
hook.pytest_runtest_logreport(report=report)
if check_interactive_exception(call, report):
@ -200,15 +207,17 @@ def call_and_report(
return report
def check_interactive_exception(call, report):
return call.excinfo and not (
def check_interactive_exception(call: "CallInfo", report: BaseReport) -> bool:
return call.excinfo is not None and not (
hasattr(report, "wasxfail")
or call.excinfo.errisinstance(Skipped)
or call.excinfo.errisinstance(bdb.BdbQuit)
)
def call_runtest_hook(item, when: "Literal['setup', 'call', 'teardown']", **kwds):
def call_runtest_hook(
item: Item, when: "Literal['setup', 'call', 'teardown']", **kwds
) -> "CallInfo":
if when == "setup":
ihook = item.ihook.pytest_runtest_setup
elif when == "call":
@ -278,13 +287,13 @@ class CallInfo:
excinfo=excinfo,
)
def __repr__(self):
def __repr__(self) -> str:
if self.excinfo is None:
return "<CallInfo when={!r} result: {!r}>".format(self.when, self._result)
return "<CallInfo when={!r} excinfo={!r}>".format(self.when, self.excinfo)
def pytest_runtest_makereport(item, call):
def pytest_runtest_makereport(item: Item, call: CallInfo) -> TestReport:
return TestReport.from_item_and_call(item, call)

View File

@ -3,9 +3,12 @@ from _pytest.config import Config
from _pytest.config import hookimpl
from _pytest.config.argparsing import Parser
from _pytest.mark.evaluate import MarkEvaluator
from _pytest.nodes import Item
from _pytest.outcomes import fail
from _pytest.outcomes import skip
from _pytest.outcomes import xfail
from _pytest.python import Function
from _pytest.runner import CallInfo
from _pytest.store import StoreKey
@ -74,7 +77,7 @@ def pytest_configure(config: Config) -> None:
@hookimpl(tryfirst=True)
def pytest_runtest_setup(item):
def pytest_runtest_setup(item: Item) -> None:
# Check if skip or skipif are specified as pytest marks
item._store[skipped_by_mark_key] = False
eval_skipif = MarkEvaluator(item, "skipif")
@ -96,7 +99,7 @@ def pytest_runtest_setup(item):
@hookimpl(hookwrapper=True)
def pytest_pyfunc_call(pyfuncitem):
def pytest_pyfunc_call(pyfuncitem: Function):
check_xfail_no_run(pyfuncitem)
outcome = yield
passed = outcome.excinfo is None
@ -104,7 +107,7 @@ def pytest_pyfunc_call(pyfuncitem):
check_strict_xfail(pyfuncitem)
def check_xfail_no_run(item):
def check_xfail_no_run(item: Item) -> None:
"""check xfail(run=False)"""
if not item.config.option.runxfail:
evalxfail = item._store[evalxfail_key]
@ -113,7 +116,7 @@ def check_xfail_no_run(item):
xfail("[NOTRUN] " + evalxfail.getexplanation())
def check_strict_xfail(pyfuncitem):
def check_strict_xfail(pyfuncitem: Function) -> None:
"""check xfail(strict=True) for the given PASSING test"""
evalxfail = pyfuncitem._store[evalxfail_key]
if evalxfail.istrue():
@ -126,7 +129,7 @@ def check_strict_xfail(pyfuncitem):
@hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
def pytest_runtest_makereport(item: Item, call: CallInfo):
outcome = yield
rep = outcome.get_result()
evalxfail = item._store.get(evalxfail_key, None)
@ -171,6 +174,7 @@ def pytest_runtest_makereport(item, call):
# the location of where the skip exception was raised within pytest
_, _, reason = rep.longrepr
filename, line = item.reportinfo()[:2]
assert line is not None
rep.longrepr = str(filename), line + 1, reason

View File

@ -2,6 +2,7 @@ import pytest
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.main import Session
from _pytest.reports import TestReport
def pytest_addoption(parser: Parser) -> None:
@ -73,7 +74,7 @@ class StepwisePlugin:
config.hook.pytest_deselected(items=already_passed)
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: TestReport) -> None:
if not self.active:
return

View File

@ -12,6 +12,7 @@ from functools import partial
from typing import Any
from typing import Callable
from typing import Dict
from typing import Generator
from typing import List
from typing import Mapping
from typing import Optional
@ -30,15 +31,19 @@ from _pytest import timing
from _pytest._io import TerminalWriter
from _pytest._io.wcwidth import wcswidth
from _pytest.compat import order_preserving_dict
from _pytest.compat import TYPE_CHECKING
from _pytest.config import _PluggyPlugin
from _pytest.config import Config
from _pytest.config import ExitCode
from _pytest.config.argparsing import Parser
from _pytest.deprecated import TERMINALWRITER_WRITER
from _pytest.main import Session
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
if TYPE_CHECKING:
from _pytest.main import Session
REPORT_COLLECTING_RESOLUTION = 0.5
KNOWN_TYPES = (
@ -610,7 +615,7 @@ class TerminalReporter:
self.write_line(line)
@pytest.hookimpl(trylast=True)
def pytest_sessionstart(self, session: Session) -> None:
def pytest_sessionstart(self, session: "Session") -> None:
self._session = session
self._sessionstarttime = timing.time()
if not self.showheader:
@ -720,7 +725,9 @@ class TerminalReporter:
self._tw.line("{}{}".format(indent + " ", line))
@pytest.hookimpl(hookwrapper=True)
def pytest_sessionfinish(self, session: Session, exitstatus: Union[int, ExitCode]):
def pytest_sessionfinish(
self, session: "Session", exitstatus: Union[int, ExitCode]
):
outcome = yield
outcome.get_result()
self._tw.line("")
@ -745,7 +752,7 @@ class TerminalReporter:
self.summary_stats()
@pytest.hookimpl(hookwrapper=True)
def pytest_terminal_summary(self):
def pytest_terminal_summary(self) -> Generator[None, None, None]:
self.summary_errors()
self.summary_failures()
self.summary_warnings()

View File

@ -1,6 +1,8 @@
""" discovery and running of std-library "unittest" style tests. """
import sys
import traceback
from typing import Any
from typing import Generator
from typing import Iterable
from typing import Optional
from typing import Union
@ -253,7 +255,7 @@ class TestCaseFunction(Function):
@hookimpl(tryfirst=True)
def pytest_runtest_makereport(item, call):
def pytest_runtest_makereport(item: Item, call: CallInfo) -> None:
if isinstance(item, TestCaseFunction):
if item._excinfo:
call.excinfo = item._excinfo.pop(0)
@ -263,7 +265,13 @@ def pytest_runtest_makereport(item, call):
pass
unittest = sys.modules.get("unittest")
if unittest and call.excinfo and call.excinfo.errisinstance(unittest.SkipTest):
if (
unittest
and call.excinfo
and call.excinfo.errisinstance(
unittest.SkipTest # type: ignore[attr-defined] # noqa: F821
)
):
# let's substitute the excinfo with a pytest.skip one
call2 = CallInfo.from_call(
lambda: pytest.skip(str(call.excinfo.value)), call.when
@ -275,9 +283,9 @@ def pytest_runtest_makereport(item, call):
@hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item):
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
if isinstance(item, TestCaseFunction) and "twisted.trial.unittest" in sys.modules:
ut = sys.modules["twisted.python.failure"]
ut = sys.modules["twisted.python.failure"] # type: Any
Failure__init__ = ut.Failure.__init__
check_testcase_implements_trial_reporter()

View File

@ -11,6 +11,8 @@ from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.main import Session
from _pytest.nodes import Item
from _pytest.terminal import TerminalReporter
if TYPE_CHECKING:
from typing_extensions import Type
@ -145,7 +147,7 @@ def warning_record_to_str(warning_message):
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_protocol(item):
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
with catch_warnings_for_item(
config=item.config, ihook=item.ihook, when="runtest", item=item
):
@ -162,7 +164,9 @@ def pytest_collection(session: Session) -> Generator[None, None, None]:
@pytest.hookimpl(hookwrapper=True)
def pytest_terminal_summary(terminalreporter):
def pytest_terminal_summary(
terminalreporter: TerminalReporter,
) -> Generator[None, None, None]:
config = terminalreporter.config
with catch_warnings_for_item(
config=config, ihook=config.hook, when="config", item=None

View File

@ -884,7 +884,7 @@ def test_store_except_info_on_error() -> None:
raise IndexError("TEST")
try:
runner.pytest_runtest_call(ItemMightRaise())
runner.pytest_runtest_call(ItemMightRaise()) # type: ignore[arg-type] # noqa: F821
except IndexError:
pass
# Check that exception info is stored on sys
@ -895,7 +895,7 @@ def test_store_except_info_on_error() -> None:
# The next run should clear the exception info stored by the previous run
ItemMightRaise.raise_error = False
runner.pytest_runtest_call(ItemMightRaise())
runner.pytest_runtest_call(ItemMightRaise()) # type: ignore[arg-type] # noqa: F821
assert not hasattr(sys, "last_type")
assert not hasattr(sys, "last_value")
assert not hasattr(sys, "last_traceback")