Merge pull request #7601 from bluetech/typing-longrepr

typing: resultlog, pytester, longrepr
This commit is contained in:
Ran Benita 2020-08-04 23:23:56 +03:00 committed by GitHub
commit 2bd0d97fcc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 215 additions and 102 deletions

View File

@ -25,6 +25,7 @@ from _pytest import deprecated
from _pytest import nodes from _pytest import nodes
from _pytest import timing from _pytest import timing
from _pytest._code.code import ExceptionRepr from _pytest._code.code import ExceptionRepr
from _pytest._code.code import ReprFileLocation
from _pytest.config import Config from _pytest.config import Config
from _pytest.config import filename_arg from _pytest.config import filename_arg
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
@ -200,8 +201,11 @@ class _NodeReporter:
self._add_simple("skipped", "xfail-marked test passes unexpectedly") self._add_simple("skipped", "xfail-marked test passes unexpectedly")
else: else:
assert report.longrepr is not None assert report.longrepr is not None
if getattr(report.longrepr, "reprcrash", None) is not None: reprcrash = getattr(
message = report.longrepr.reprcrash.message report.longrepr, "reprcrash", None
) # type: Optional[ReprFileLocation]
if reprcrash is not None:
message = reprcrash.message
else: else:
message = str(report.longrepr) message = str(report.longrepr)
message = bin_xml_escape(message) message = bin_xml_escape(message)
@ -217,8 +221,11 @@ class _NodeReporter:
def append_error(self, report: TestReport) -> None: def append_error(self, report: TestReport) -> None:
assert report.longrepr is not None assert report.longrepr is not None
if getattr(report.longrepr, "reprcrash", None) is not None: reprcrash = getattr(
reason = report.longrepr.reprcrash.message report.longrepr, "reprcrash", None
) # type: Optional[ReprFileLocation]
if reprcrash is not None:
reason = reprcrash.message
else: else:
reason = str(report.longrepr) reason = str(report.longrepr)
@ -237,7 +244,7 @@ class _NodeReporter:
skipped = ET.Element("skipped", type="pytest.xfail", message=xfailreason) skipped = ET.Element("skipped", type="pytest.xfail", message=xfailreason)
self.append(skipped) self.append(skipped)
else: else:
assert report.longrepr is not None assert isinstance(report.longrepr, tuple)
filename, lineno, skipreason = report.longrepr filename, lineno, skipreason = report.longrepr
if skipreason.startswith("Skipped: "): if skipreason.startswith("Skipped: "):
skipreason = skipreason[9:] skipreason = skipreason[9:]

View File

@ -28,6 +28,7 @@ import pytest
from _pytest import timing from _pytest import timing
from _pytest._code import Source from _pytest._code import Source
from _pytest.capture import _get_multicapture from _pytest.capture import _get_multicapture
from _pytest.compat import overload
from _pytest.compat import TYPE_CHECKING from _pytest.compat import TYPE_CHECKING
from _pytest.config import _PluggyPlugin from _pytest.config import _PluggyPlugin
from _pytest.config import Config from _pytest.config import Config
@ -42,11 +43,13 @@ from _pytest.nodes import Item
from _pytest.pathlib import make_numbered_dir from _pytest.pathlib import make_numbered_dir
from _pytest.pathlib import Path from _pytest.pathlib import Path
from _pytest.python import Module from _pytest.python import Module
from _pytest.reports import CollectReport
from _pytest.reports import TestReport from _pytest.reports import TestReport
from _pytest.tmpdir import TempdirFactory from _pytest.tmpdir import TempdirFactory
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Type from typing import Type
from typing_extensions import Literal
import pexpect import pexpect
@ -180,24 +183,24 @@ class PytestArg:
return hookrecorder return hookrecorder
def get_public_names(values): def get_public_names(values: Iterable[str]) -> List[str]:
"""Only return names from iterator values without a leading underscore.""" """Only return names from iterator values without a leading underscore."""
return [x for x in values if x[0] != "_"] return [x for x in values if x[0] != "_"]
class ParsedCall: class ParsedCall:
def __init__(self, name, kwargs): def __init__(self, name: str, kwargs) -> None:
self.__dict__.update(kwargs) self.__dict__.update(kwargs)
self._name = name self._name = name
def __repr__(self): def __repr__(self) -> str:
d = self.__dict__.copy() d = self.__dict__.copy()
del d["_name"] del d["_name"]
return "<ParsedCall {!r}(**{!r})>".format(self._name, d) return "<ParsedCall {!r}(**{!r})>".format(self._name, d)
if TYPE_CHECKING: if TYPE_CHECKING:
# The class has undetermined attributes, this tells mypy about it. # The class has undetermined attributes, this tells mypy about it.
def __getattr__(self, key): def __getattr__(self, key: str):
raise NotImplementedError() raise NotImplementedError()
@ -211,6 +214,7 @@ class HookRecorder:
def __init__(self, pluginmanager: PytestPluginManager) -> None: def __init__(self, pluginmanager: PytestPluginManager) -> None:
self._pluginmanager = pluginmanager self._pluginmanager = pluginmanager
self.calls = [] # type: List[ParsedCall] self.calls = [] # type: List[ParsedCall]
self.ret = None # type: Optional[Union[int, ExitCode]]
def before(hook_name: str, hook_impls, kwargs) -> None: def before(hook_name: str, hook_impls, kwargs) -> None:
self.calls.append(ParsedCall(hook_name, kwargs)) self.calls.append(ParsedCall(hook_name, kwargs))
@ -228,7 +232,7 @@ class HookRecorder:
names = names.split() names = names.split()
return [call for call in self.calls if call._name in names] return [call for call in self.calls if call._name in names]
def assert_contains(self, entries) -> None: def assert_contains(self, entries: Sequence[Tuple[str, str]]) -> None:
__tracebackhide__ = True __tracebackhide__ = True
i = 0 i = 0
entries = list(entries) entries = list(entries)
@ -266,22 +270,46 @@ class HookRecorder:
# functionality for test reports # functionality for test reports
@overload
def getreports( def getreports(
self, names: "Literal['pytest_collectreport']",
) -> Sequence[CollectReport]:
raise NotImplementedError()
@overload # noqa: F811
def getreports( # noqa: F811
self, names: "Literal['pytest_runtest_logreport']",
) -> Sequence[TestReport]:
raise NotImplementedError()
@overload # noqa: F811
def getreports( # noqa: F811
self, self,
names: Union[ names: Union[str, Iterable[str]] = (
str, Iterable[str] "pytest_collectreport",
] = "pytest_runtest_logreport pytest_collectreport", "pytest_runtest_logreport",
) -> List[TestReport]: ),
) -> Sequence[Union[CollectReport, TestReport]]:
raise NotImplementedError()
def getreports( # noqa: F811
self,
names: Union[str, Iterable[str]] = (
"pytest_collectreport",
"pytest_runtest_logreport",
),
) -> Sequence[Union[CollectReport, TestReport]]:
return [x.report for x in self.getcalls(names)] return [x.report for x in self.getcalls(names)]
def matchreport( def matchreport(
self, self,
inamepart: str = "", inamepart: str = "",
names: Union[ names: Union[str, Iterable[str]] = (
str, Iterable[str] "pytest_runtest_logreport",
] = "pytest_runtest_logreport pytest_collectreport", "pytest_collectreport",
when=None, ),
): when: Optional[str] = None,
) -> Union[CollectReport, TestReport]:
"""Return a testreport whose dotted import path matches.""" """Return a testreport whose dotted import path matches."""
values = [] values = []
for rep in self.getreports(names=names): for rep in self.getreports(names=names):
@ -305,26 +333,56 @@ class HookRecorder:
) )
return values[0] return values[0]
@overload
def getfailures( def getfailures(
self, names: "Literal['pytest_collectreport']",
) -> Sequence[CollectReport]:
raise NotImplementedError()
@overload # noqa: F811
def getfailures( # noqa: F811
self, names: "Literal['pytest_runtest_logreport']",
) -> Sequence[TestReport]:
raise NotImplementedError()
@overload # noqa: F811
def getfailures( # noqa: F811
self, self,
names: Union[ names: Union[str, Iterable[str]] = (
str, Iterable[str] "pytest_collectreport",
] = "pytest_runtest_logreport pytest_collectreport", "pytest_runtest_logreport",
) -> List[TestReport]: ),
) -> Sequence[Union[CollectReport, TestReport]]:
raise NotImplementedError()
def getfailures( # noqa: F811
self,
names: Union[str, Iterable[str]] = (
"pytest_collectreport",
"pytest_runtest_logreport",
),
) -> Sequence[Union[CollectReport, TestReport]]:
return [rep for rep in self.getreports(names) if rep.failed] return [rep for rep in self.getreports(names) if rep.failed]
def getfailedcollections(self) -> List[TestReport]: def getfailedcollections(self) -> Sequence[CollectReport]:
return self.getfailures("pytest_collectreport") return self.getfailures("pytest_collectreport")
def listoutcomes( def listoutcomes(
self, self,
) -> Tuple[List[TestReport], List[TestReport], List[TestReport]]: ) -> Tuple[
Sequence[TestReport],
Sequence[Union[CollectReport, TestReport]],
Sequence[Union[CollectReport, TestReport]],
]:
passed = [] passed = []
skipped = [] skipped = []
failed = [] failed = []
for rep in self.getreports("pytest_collectreport pytest_runtest_logreport"): for rep in self.getreports(
("pytest_collectreport", "pytest_runtest_logreport")
):
if rep.passed: if rep.passed:
if rep.when == "call": if rep.when == "call":
assert isinstance(rep, TestReport)
passed.append(rep) passed.append(rep)
elif rep.skipped: elif rep.skipped:
skipped.append(rep) skipped.append(rep)
@ -879,7 +937,7 @@ class Testdir:
runner = testclassinstance.getrunner() runner = testclassinstance.getrunner()
return runner(item) return runner(item)
def inline_runsource(self, source, *cmdlineargs): def inline_runsource(self, source, *cmdlineargs) -> HookRecorder:
"""Run a test module in process using ``pytest.main()``. """Run a test module in process using ``pytest.main()``.
This run writes "source" into a temporary file and runs This run writes "source" into a temporary file and runs
@ -896,7 +954,7 @@ class Testdir:
values = list(cmdlineargs) + [p] values = list(cmdlineargs) + [p]
return self.inline_run(*values) return self.inline_run(*values)
def inline_genitems(self, *args): def inline_genitems(self, *args) -> Tuple[List[Item], HookRecorder]:
"""Run ``pytest.main(['--collectonly'])`` in-process. """Run ``pytest.main(['--collectonly'])`` in-process.
Runs the :py:func:`pytest.main` function to run all of pytest inside Runs the :py:func:`pytest.main` function to run all of pytest inside
@ -907,7 +965,9 @@ class Testdir:
items = [x.item for x in rec.getcalls("pytest_itemcollected")] items = [x.item for x in rec.getcalls("pytest_itemcollected")]
return items, rec return items, rec
def inline_run(self, *args, plugins=(), no_reraise_ctrlc: bool = False): def inline_run(
self, *args, plugins=(), no_reraise_ctrlc: bool = False
) -> HookRecorder:
"""Run ``pytest.main()`` in-process, returning a HookRecorder. """Run ``pytest.main()`` in-process, returning a HookRecorder.
Runs the :py:func:`pytest.main` function to run all of pytest inside Runs the :py:func:`pytest.main` function to run all of pytest inside
@ -962,7 +1022,7 @@ class Testdir:
class reprec: # type: ignore class reprec: # type: ignore
pass pass
reprec.ret = ret # type: ignore[attr-defined] reprec.ret = ret
# Typically we reraise keyboard interrupts from the child run # Typically we reraise keyboard interrupts from the child run
# because it's our user requesting interruption of the testing. # because it's our user requesting interruption of the testing.
@ -1010,6 +1070,7 @@ class Testdir:
sys.stdout.write(out) sys.stdout.write(out)
sys.stderr.write(err) sys.stderr.write(err)
assert reprec.ret is not None
res = RunResult( res = RunResult(
reprec.ret, out.splitlines(), err.splitlines(), timing.time() - now reprec.ret, out.splitlines(), err.splitlines(), timing.time() - now
) )

View File

@ -1,6 +1,7 @@
from io import StringIO from io import StringIO
from pprint import pprint from pprint import pprint
from typing import Any from typing import Any
from typing import cast
from typing import Dict from typing import Dict
from typing import Iterable from typing import Iterable
from typing import Iterator from typing import Iterator
@ -15,6 +16,7 @@ import py
from _pytest._code.code import ExceptionChainRepr from _pytest._code.code import ExceptionChainRepr
from _pytest._code.code import ExceptionInfo from _pytest._code.code import ExceptionInfo
from _pytest._code.code import ExceptionRepr
from _pytest._code.code import ReprEntry from _pytest._code.code import ReprEntry
from _pytest._code.code import ReprEntryNative from _pytest._code.code import ReprEntryNative
from _pytest._code.code import ReprExceptionInfo from _pytest._code.code import ReprExceptionInfo
@ -57,8 +59,9 @@ _R = TypeVar("_R", bound="BaseReport")
class BaseReport: class BaseReport:
when = None # type: Optional[str] when = None # type: Optional[str]
location = None # type: Optional[Tuple[str, Optional[int], str]] location = None # type: Optional[Tuple[str, Optional[int], str]]
# TODO: Improve this Any. longrepr = (
longrepr = None # type: Optional[Any] None
) # type: Union[None, ExceptionInfo[BaseException], Tuple[str, int, str], str, TerminalRepr]
sections = [] # type: List[Tuple[str, str]] sections = [] # type: List[Tuple[str, str]]
nodeid = None # type: str nodeid = None # type: str
@ -79,7 +82,8 @@ class BaseReport:
return return
if hasattr(longrepr, "toterminal"): if hasattr(longrepr, "toterminal"):
longrepr.toterminal(out) longrepr_terminal = cast(TerminalRepr, longrepr)
longrepr_terminal.toterminal(out)
else: else:
try: try:
s = str(longrepr) s = str(longrepr)
@ -233,7 +237,9 @@ class TestReport(BaseReport):
location: Tuple[str, Optional[int], str], location: Tuple[str, Optional[int], str],
keywords, keywords,
outcome: "Literal['passed', 'failed', 'skipped']", outcome: "Literal['passed', 'failed', 'skipped']",
longrepr, longrepr: Union[
None, ExceptionInfo[BaseException], Tuple[str, int, str], str, TerminalRepr
],
when: "Literal['setup', 'call', 'teardown']", when: "Literal['setup', 'call', 'teardown']",
sections: Iterable[Tuple[str, str]] = (), sections: Iterable[Tuple[str, str]] = (),
duration: float = 0, duration: float = 0,
@ -293,8 +299,9 @@ class TestReport(BaseReport):
sections = [] sections = []
if not call.excinfo: if not call.excinfo:
outcome = "passed" # type: Literal["passed", "failed", "skipped"] outcome = "passed" # type: Literal["passed", "failed", "skipped"]
# TODO: Improve this Any. longrepr = (
longrepr = None # type: Optional[Any] None
) # type: Union[None, ExceptionInfo[BaseException], Tuple[str, int, str], str, TerminalRepr]
else: else:
if not isinstance(excinfo, ExceptionInfo): if not isinstance(excinfo, ExceptionInfo):
outcome = "failed" outcome = "failed"
@ -372,7 +379,7 @@ class CollectReport(BaseReport):
class CollectErrorRepr(TerminalRepr): class CollectErrorRepr(TerminalRepr):
def __init__(self, msg) -> None: def __init__(self, msg: str) -> None:
self.longrepr = msg self.longrepr = msg
def toterminal(self, out: TerminalWriter) -> None: def toterminal(self, out: TerminalWriter) -> None:
@ -436,16 +443,18 @@ def _report_to_json(report: BaseReport) -> Dict[str, Any]:
else: else:
return None return None
def serialize_longrepr(rep: BaseReport) -> Dict[str, Any]: def serialize_exception_longrepr(rep: BaseReport) -> Dict[str, Any]:
assert rep.longrepr is not None assert rep.longrepr is not None
# TODO: Investigate whether the duck typing is really necessary here.
longrepr = cast(ExceptionRepr, rep.longrepr)
result = { result = {
"reprcrash": serialize_repr_crash(rep.longrepr.reprcrash), "reprcrash": serialize_repr_crash(longrepr.reprcrash),
"reprtraceback": serialize_repr_traceback(rep.longrepr.reprtraceback), "reprtraceback": serialize_repr_traceback(longrepr.reprtraceback),
"sections": rep.longrepr.sections, "sections": longrepr.sections,
} # type: Dict[str, Any] } # type: Dict[str, Any]
if isinstance(rep.longrepr, ExceptionChainRepr): if isinstance(longrepr, ExceptionChainRepr):
result["chain"] = [] result["chain"] = []
for repr_traceback, repr_crash, description in rep.longrepr.chain: for repr_traceback, repr_crash, description in longrepr.chain:
result["chain"].append( result["chain"].append(
( (
serialize_repr_traceback(repr_traceback), serialize_repr_traceback(repr_traceback),
@ -462,7 +471,7 @@ def _report_to_json(report: BaseReport) -> Dict[str, Any]:
if hasattr(report.longrepr, "reprtraceback") and hasattr( if hasattr(report.longrepr, "reprtraceback") and hasattr(
report.longrepr, "reprcrash" report.longrepr, "reprcrash"
): ):
d["longrepr"] = serialize_longrepr(report) d["longrepr"] = serialize_exception_longrepr(report)
else: else:
d["longrepr"] = str(report.longrepr) d["longrepr"] = str(report.longrepr)
else: else:

View File

@ -1,5 +1,7 @@
"""log machine-parseable test session result information to a plain text file.""" """log machine-parseable test session result information to a plain text file."""
import os import os
from typing import IO
from typing import Union
import py import py
@ -52,16 +54,18 @@ def pytest_unconfigure(config: Config) -> None:
class ResultLog: class ResultLog:
def __init__(self, config, logfile): def __init__(self, config: Config, logfile: IO[str]) -> None:
self.config = config self.config = config
self.logfile = logfile # preferably line buffered self.logfile = logfile # preferably line buffered
def write_log_entry(self, testpath, lettercode, longrepr): def write_log_entry(self, testpath: str, lettercode: str, longrepr: str) -> None:
print("{} {}".format(lettercode, testpath), file=self.logfile) print("{} {}".format(lettercode, testpath), file=self.logfile)
for line in longrepr.splitlines(): for line in longrepr.splitlines():
print(" %s" % line, file=self.logfile) print(" %s" % line, file=self.logfile)
def log_outcome(self, report, lettercode, longrepr): def log_outcome(
self, report: Union[TestReport, CollectReport], lettercode: str, longrepr: str
) -> None:
testpath = getattr(report, "nodeid", None) testpath = getattr(report, "nodeid", None)
if testpath is None: if testpath is None:
testpath = report.fspath testpath = report.fspath
@ -73,7 +77,7 @@ class ResultLog:
res = self.config.hook.pytest_report_teststatus( res = self.config.hook.pytest_report_teststatus(
report=report, config=self.config report=report, config=self.config
) )
code = res[1] code = res[1] # type: str
if code == "x": if code == "x":
longrepr = str(report.longrepr) longrepr = str(report.longrepr)
elif code == "X": elif code == "X":
@ -81,7 +85,7 @@ class ResultLog:
elif report.passed: elif report.passed:
longrepr = "" longrepr = ""
elif report.skipped: elif report.skipped:
assert report.longrepr is not None assert isinstance(report.longrepr, tuple)
longrepr = str(report.longrepr[2]) longrepr = str(report.longrepr[2])
else: else:
longrepr = str(report.longrepr) longrepr = str(report.longrepr)

View File

@ -2,7 +2,6 @@
import bdb import bdb
import os import os
import sys import sys
from typing import Any
from typing import Callable from typing import Callable
from typing import cast from typing import cast
from typing import Dict from typing import Dict
@ -22,6 +21,7 @@ from .reports import TestReport
from _pytest import timing from _pytest import timing
from _pytest._code.code import ExceptionChainRepr from _pytest._code.code import ExceptionChainRepr
from _pytest._code.code import ExceptionInfo from _pytest._code.code import ExceptionInfo
from _pytest._code.code import TerminalRepr
from _pytest.compat import TYPE_CHECKING from _pytest.compat import TYPE_CHECKING
from _pytest.config.argparsing import Parser from _pytest.config.argparsing import Parser
from _pytest.nodes import Collector from _pytest.nodes import Collector
@ -327,8 +327,7 @@ def pytest_runtest_makereport(item: Item, call: CallInfo[None]) -> TestReport:
def pytest_make_collect_report(collector: Collector) -> CollectReport: def pytest_make_collect_report(collector: Collector) -> CollectReport:
call = CallInfo.from_call(lambda: list(collector.collect()), "collect") call = CallInfo.from_call(lambda: list(collector.collect()), "collect")
# TODO: Better typing for longrepr. longrepr = None # type: Union[None, Tuple[str, int, str], str, TerminalRepr]
longrepr = None # type: Optional[Any]
if not call.excinfo: if not call.excinfo:
outcome = "passed" # type: Literal["passed", "skipped", "failed"] outcome = "passed" # type: Literal["passed", "skipped", "failed"]
else: else:
@ -348,6 +347,7 @@ def pytest_make_collect_report(collector: Collector) -> CollectReport:
outcome = "failed" outcome = "failed"
errorinfo = collector.repr_failure(call.excinfo) errorinfo = collector.repr_failure(call.excinfo)
if not hasattr(errorinfo, "toterminal"): if not hasattr(errorinfo, "toterminal"):
assert isinstance(errorinfo, str)
errorinfo = CollectErrorRepr(errorinfo) errorinfo = CollectErrorRepr(errorinfo)
longrepr = errorinfo longrepr = errorinfo
result = call.result if not call.excinfo else None result = call.result if not call.excinfo else None

View File

@ -1247,6 +1247,7 @@ def _folded_skips(
d = {} # type: Dict[Tuple[str, Optional[int], str], List[CollectReport]] d = {} # type: Dict[Tuple[str, Optional[int], str], List[CollectReport]]
for event in skipped: for event in skipped:
assert event.longrepr is not None assert event.longrepr is not None
assert isinstance(event.longrepr, tuple), (event, event.longrepr)
assert len(event.longrepr) == 3, (event, event.longrepr) assert len(event.longrepr) == 3, (event, event.longrepr)
fspath, lineno, reason = event.longrepr fspath, lineno, reason = event.longrepr
# For consistency, report all fspaths in relative form. # For consistency, report all fspaths in relative form.

View File

@ -1,14 +1,19 @@
import sys import sys
from typing import Sequence
from typing import Union
import pytest import pytest
from _pytest._code.code import ExceptionChainRepr from _pytest._code.code import ExceptionChainRepr
from _pytest._code.code import ExceptionRepr
from _pytest.config import Config
from _pytest.pathlib import Path from _pytest.pathlib import Path
from _pytest.pytester import Testdir
from _pytest.reports import CollectReport from _pytest.reports import CollectReport
from _pytest.reports import TestReport from _pytest.reports import TestReport
class TestReportSerialization: class TestReportSerialization:
def test_xdist_longrepr_to_str_issue_241(self, testdir): def test_xdist_longrepr_to_str_issue_241(self, testdir: Testdir) -> None:
"""Regarding issue pytest-xdist#241. """Regarding issue pytest-xdist#241.
This test came originally from test_remote.py in xdist (ca03269). This test came originally from test_remote.py in xdist (ca03269).
@ -31,7 +36,7 @@ class TestReportSerialization:
assert test_b_call.outcome == "passed" assert test_b_call.outcome == "passed"
assert test_b_call._to_json()["longrepr"] is None assert test_b_call._to_json()["longrepr"] is None
def test_xdist_report_longrepr_reprcrash_130(self, testdir) -> None: def test_xdist_report_longrepr_reprcrash_130(self, testdir: Testdir) -> None:
"""Regarding issue pytest-xdist#130 """Regarding issue pytest-xdist#130
This test came originally from test_remote.py in xdist (ca03269). This test came originally from test_remote.py in xdist (ca03269).
@ -46,15 +51,18 @@ class TestReportSerialization:
assert len(reports) == 3 assert len(reports) == 3
rep = reports[1] rep = reports[1]
added_section = ("Failure Metadata", "metadata metadata", "*") added_section = ("Failure Metadata", "metadata metadata", "*")
assert isinstance(rep.longrepr, ExceptionRepr)
rep.longrepr.sections.append(added_section) rep.longrepr.sections.append(added_section)
d = rep._to_json() d = rep._to_json()
a = TestReport._from_json(d) a = TestReport._from_json(d)
assert a.longrepr is not None assert isinstance(a.longrepr, ExceptionRepr)
# Check assembled == rep # Check assembled == rep
assert a.__dict__.keys() == rep.__dict__.keys() assert a.__dict__.keys() == rep.__dict__.keys()
for key in rep.__dict__.keys(): for key in rep.__dict__.keys():
if key != "longrepr": if key != "longrepr":
assert getattr(a, key) == getattr(rep, key) assert getattr(a, key) == getattr(rep, key)
assert rep.longrepr.reprcrash is not None
assert a.longrepr.reprcrash is not None
assert rep.longrepr.reprcrash.lineno == a.longrepr.reprcrash.lineno assert rep.longrepr.reprcrash.lineno == a.longrepr.reprcrash.lineno
assert rep.longrepr.reprcrash.message == a.longrepr.reprcrash.message assert rep.longrepr.reprcrash.message == a.longrepr.reprcrash.message
assert rep.longrepr.reprcrash.path == a.longrepr.reprcrash.path assert rep.longrepr.reprcrash.path == a.longrepr.reprcrash.path
@ -67,7 +75,7 @@ class TestReportSerialization:
# Missing section attribute PR171 # Missing section attribute PR171
assert added_section in a.longrepr.sections assert added_section in a.longrepr.sections
def test_reprentries_serialization_170(self, testdir) -> None: def test_reprentries_serialization_170(self, testdir: Testdir) -> None:
"""Regarding issue pytest-xdist#170 """Regarding issue pytest-xdist#170
This test came originally from test_remote.py in xdist (ca03269). This test came originally from test_remote.py in xdist (ca03269).
@ -85,25 +93,35 @@ class TestReportSerialization:
reports = reprec.getreports("pytest_runtest_logreport") reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3 assert len(reports) == 3
rep = reports[1] rep = reports[1]
assert isinstance(rep.longrepr, ExceptionRepr)
d = rep._to_json() d = rep._to_json()
a = TestReport._from_json(d) a = TestReport._from_json(d)
assert a.longrepr is not None assert isinstance(a.longrepr, ExceptionRepr)
rep_entries = rep.longrepr.reprtraceback.reprentries rep_entries = rep.longrepr.reprtraceback.reprentries
a_entries = a.longrepr.reprtraceback.reprentries a_entries = a.longrepr.reprtraceback.reprentries
for i in range(len(a_entries)): for i in range(len(a_entries)):
assert isinstance(rep_entries[i], ReprEntry) rep_entry = rep_entries[i]
assert rep_entries[i].lines == a_entries[i].lines assert isinstance(rep_entry, ReprEntry)
assert rep_entries[i].reprfileloc.lineno == a_entries[i].reprfileloc.lineno assert rep_entry.reprfileloc is not None
assert ( assert rep_entry.reprfuncargs is not None
rep_entries[i].reprfileloc.message == a_entries[i].reprfileloc.message assert rep_entry.reprlocals is not None
)
assert rep_entries[i].reprfileloc.path == a_entries[i].reprfileloc.path
assert rep_entries[i].reprfuncargs.args == a_entries[i].reprfuncargs.args
assert rep_entries[i].reprlocals.lines == a_entries[i].reprlocals.lines
assert rep_entries[i].style == a_entries[i].style
def test_reprentries_serialization_196(self, testdir) -> None: a_entry = a_entries[i]
assert isinstance(a_entry, ReprEntry)
assert a_entry.reprfileloc is not None
assert a_entry.reprfuncargs is not None
assert a_entry.reprlocals is not None
assert rep_entry.lines == a_entry.lines
assert rep_entry.reprfileloc.lineno == a_entry.reprfileloc.lineno
assert rep_entry.reprfileloc.message == a_entry.reprfileloc.message
assert rep_entry.reprfileloc.path == a_entry.reprfileloc.path
assert rep_entry.reprfuncargs.args == a_entry.reprfuncargs.args
assert rep_entry.reprlocals.lines == a_entry.reprlocals.lines
assert rep_entry.style == a_entry.style
def test_reprentries_serialization_196(self, testdir: Testdir) -> None:
"""Regarding issue pytest-xdist#196 """Regarding issue pytest-xdist#196
This test came originally from test_remote.py in xdist (ca03269). This test came originally from test_remote.py in xdist (ca03269).
@ -121,9 +139,10 @@ class TestReportSerialization:
reports = reprec.getreports("pytest_runtest_logreport") reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3 assert len(reports) == 3
rep = reports[1] rep = reports[1]
assert isinstance(rep.longrepr, ExceptionRepr)
d = rep._to_json() d = rep._to_json()
a = TestReport._from_json(d) a = TestReport._from_json(d)
assert a.longrepr is not None assert isinstance(a.longrepr, ExceptionRepr)
rep_entries = rep.longrepr.reprtraceback.reprentries rep_entries = rep.longrepr.reprtraceback.reprentries
a_entries = a.longrepr.reprtraceback.reprentries a_entries = a.longrepr.reprtraceback.reprentries
@ -131,7 +150,7 @@ class TestReportSerialization:
assert isinstance(rep_entries[i], ReprEntryNative) assert isinstance(rep_entries[i], ReprEntryNative)
assert rep_entries[i].lines == a_entries[i].lines assert rep_entries[i].lines == a_entries[i].lines
def test_itemreport_outcomes(self, testdir): def test_itemreport_outcomes(self, testdir: Testdir) -> None:
# This test came originally from test_remote.py in xdist (ca03269). # This test came originally from test_remote.py in xdist (ca03269).
reprec = testdir.inline_runsource( reprec = testdir.inline_runsource(
""" """
@ -157,7 +176,7 @@ class TestReportSerialization:
assert newrep.failed == rep.failed assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped assert newrep.skipped == rep.skipped
if newrep.skipped and not hasattr(newrep, "wasxfail"): if newrep.skipped and not hasattr(newrep, "wasxfail"):
assert newrep.longrepr is not None assert isinstance(newrep.longrepr, tuple)
assert len(newrep.longrepr) == 3 assert len(newrep.longrepr) == 3
assert newrep.outcome == rep.outcome assert newrep.outcome == rep.outcome
assert newrep.when == rep.when assert newrep.when == rep.when
@ -165,7 +184,7 @@ class TestReportSerialization:
if rep.failed: if rep.failed:
assert newrep.longreprtext == rep.longreprtext assert newrep.longreprtext == rep.longreprtext
def test_collectreport_passed(self, testdir): def test_collectreport_passed(self, testdir: Testdir) -> None:
"""This test came originally from test_remote.py in xdist (ca03269).""" """This test came originally from test_remote.py in xdist (ca03269)."""
reprec = testdir.inline_runsource("def test_func(): pass") reprec = testdir.inline_runsource("def test_func(): pass")
reports = reprec.getreports("pytest_collectreport") reports = reprec.getreports("pytest_collectreport")
@ -176,7 +195,7 @@ class TestReportSerialization:
assert newrep.failed == rep.failed assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped assert newrep.skipped == rep.skipped
def test_collectreport_fail(self, testdir): def test_collectreport_fail(self, testdir: Testdir) -> None:
"""This test came originally from test_remote.py in xdist (ca03269).""" """This test came originally from test_remote.py in xdist (ca03269)."""
reprec = testdir.inline_runsource("qwe abc") reprec = testdir.inline_runsource("qwe abc")
reports = reprec.getreports("pytest_collectreport") reports = reprec.getreports("pytest_collectreport")
@ -190,13 +209,13 @@ class TestReportSerialization:
if rep.failed: if rep.failed:
assert newrep.longrepr == str(rep.longrepr) assert newrep.longrepr == str(rep.longrepr)
def test_extended_report_deserialization(self, testdir): def test_extended_report_deserialization(self, testdir: Testdir) -> None:
"""This test came originally from test_remote.py in xdist (ca03269).""" """This test came originally from test_remote.py in xdist (ca03269)."""
reprec = testdir.inline_runsource("qwe abc") reprec = testdir.inline_runsource("qwe abc")
reports = reprec.getreports("pytest_collectreport") reports = reprec.getreports("pytest_collectreport")
assert reports assert reports
for rep in reports: for rep in reports:
rep.extra = True rep.extra = True # type: ignore[attr-defined]
d = rep._to_json() d = rep._to_json()
newrep = CollectReport._from_json(d) newrep = CollectReport._from_json(d)
assert newrep.extra assert newrep.extra
@ -206,7 +225,7 @@ class TestReportSerialization:
if rep.failed: if rep.failed:
assert newrep.longrepr == str(rep.longrepr) assert newrep.longrepr == str(rep.longrepr)
def test_paths_support(self, testdir): def test_paths_support(self, testdir: Testdir) -> None:
"""Report attributes which are py.path or pathlib objects should become strings.""" """Report attributes which are py.path or pathlib objects should become strings."""
testdir.makepyfile( testdir.makepyfile(
""" """
@ -218,13 +237,13 @@ class TestReportSerialization:
reports = reprec.getreports("pytest_runtest_logreport") reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3 assert len(reports) == 3
test_a_call = reports[1] test_a_call = reports[1]
test_a_call.path1 = testdir.tmpdir test_a_call.path1 = testdir.tmpdir # type: ignore[attr-defined]
test_a_call.path2 = Path(testdir.tmpdir) test_a_call.path2 = Path(testdir.tmpdir) # type: ignore[attr-defined]
data = test_a_call._to_json() data = test_a_call._to_json()
assert data["path1"] == str(testdir.tmpdir) assert data["path1"] == str(testdir.tmpdir)
assert data["path2"] == str(testdir.tmpdir) assert data["path2"] == str(testdir.tmpdir)
def test_deserialization_failure(self, testdir): def test_deserialization_failure(self, testdir: Testdir) -> None:
"""Check handling of failure during deserialization of report types.""" """Check handling of failure during deserialization of report types."""
testdir.makepyfile( testdir.makepyfile(
""" """
@ -247,7 +266,7 @@ class TestReportSerialization:
TestReport._from_json(data) TestReport._from_json(data)
@pytest.mark.parametrize("report_class", [TestReport, CollectReport]) @pytest.mark.parametrize("report_class", [TestReport, CollectReport])
def test_chained_exceptions(self, testdir, tw_mock, report_class): def test_chained_exceptions(self, testdir: Testdir, tw_mock, report_class) -> None:
"""Check serialization/deserialization of report objects containing chained exceptions (#5786)""" """Check serialization/deserialization of report objects containing chained exceptions (#5786)"""
testdir.makepyfile( testdir.makepyfile(
""" """
@ -267,7 +286,9 @@ class TestReportSerialization:
reprec = testdir.inline_run() reprec = testdir.inline_run()
if report_class is TestReport: if report_class is TestReport:
reports = reprec.getreports("pytest_runtest_logreport") reports = reprec.getreports(
"pytest_runtest_logreport"
) # type: Union[Sequence[TestReport], Sequence[CollectReport]]
# we have 3 reports: setup/call/teardown # we have 3 reports: setup/call/teardown
assert len(reports) == 3 assert len(reports) == 3
# get the call report # get the call report
@ -279,7 +300,7 @@ class TestReportSerialization:
assert len(reports) == 2 assert len(reports) == 2
report = reports[1] report = reports[1]
def check_longrepr(longrepr): def check_longrepr(longrepr: ExceptionChainRepr) -> None:
"""Check the attributes of the given longrepr object according to the test file. """Check the attributes of the given longrepr object according to the test file.
We can get away with testing both CollectReport and TestReport with this function because We can get away with testing both CollectReport and TestReport with this function because
@ -303,6 +324,7 @@ class TestReportSerialization:
assert report.failed assert report.failed
assert len(report.sections) == 0 assert len(report.sections) == 0
assert isinstance(report.longrepr, ExceptionChainRepr)
report.longrepr.addsection("title", "contents", "=") report.longrepr.addsection("title", "contents", "=")
check_longrepr(report.longrepr) check_longrepr(report.longrepr)
@ -317,7 +339,7 @@ class TestReportSerialization:
# elsewhere and we do check the contents of the longrepr object after loading it. # elsewhere and we do check the contents of the longrepr object after loading it.
loaded_report.longrepr.toterminal(tw_mock) loaded_report.longrepr.toterminal(tw_mock)
def test_chained_exceptions_no_reprcrash(self, testdir, tw_mock) -> None: def test_chained_exceptions_no_reprcrash(self, testdir: Testdir, tw_mock) -> None:
"""Regression test for tracebacks without a reprcrash (#5971) """Regression test for tracebacks without a reprcrash (#5971)
This happens notably on exceptions raised by multiprocess.pool: the exception transfer This happens notably on exceptions raised by multiprocess.pool: the exception transfer
@ -368,7 +390,7 @@ class TestReportSerialization:
reports = reprec.getreports("pytest_runtest_logreport") reports = reprec.getreports("pytest_runtest_logreport")
def check_longrepr(longrepr) -> None: def check_longrepr(longrepr: object) -> None:
assert isinstance(longrepr, ExceptionChainRepr) assert isinstance(longrepr, ExceptionChainRepr)
assert len(longrepr.chain) == 2 assert len(longrepr.chain) == 2
entry1, entry2 = longrepr.chain entry1, entry2 = longrepr.chain
@ -397,9 +419,12 @@ class TestReportSerialization:
# for same reasons as previous test, ensure we don't blow up here # for same reasons as previous test, ensure we don't blow up here
assert loaded_report.longrepr is not None assert loaded_report.longrepr is not None
assert isinstance(loaded_report.longrepr, ExceptionChainRepr)
loaded_report.longrepr.toterminal(tw_mock) loaded_report.longrepr.toterminal(tw_mock)
def test_report_prevent_ConftestImportFailure_hiding_exception(self, testdir): def test_report_prevent_ConftestImportFailure_hiding_exception(
self, testdir: Testdir
) -> None:
sub_dir = testdir.tmpdir.join("ns").ensure_dir() sub_dir = testdir.tmpdir.join("ns").ensure_dir()
sub_dir.join("conftest").new(ext=".py").write("import unknown") sub_dir.join("conftest").new(ext=".py").write("import unknown")
@ -411,7 +436,7 @@ class TestReportSerialization:
class TestHooks: class TestHooks:
"""Test that the hooks are working correctly for plugins""" """Test that the hooks are working correctly for plugins"""
def test_test_report(self, testdir, pytestconfig): def test_test_report(self, testdir: Testdir, pytestconfig: Config) -> None:
testdir.makepyfile( testdir.makepyfile(
""" """
def test_a(): assert False def test_a(): assert False
@ -433,7 +458,7 @@ class TestHooks:
assert new_rep.when == rep.when assert new_rep.when == rep.when
assert new_rep.outcome == rep.outcome assert new_rep.outcome == rep.outcome
def test_collect_report(self, testdir, pytestconfig): def test_collect_report(self, testdir: Testdir, pytestconfig: Config) -> None:
testdir.makepyfile( testdir.makepyfile(
""" """
def test_a(): assert False def test_a(): assert False
@ -458,7 +483,9 @@ class TestHooks:
@pytest.mark.parametrize( @pytest.mark.parametrize(
"hook_name", ["pytest_runtest_logreport", "pytest_collectreport"] "hook_name", ["pytest_runtest_logreport", "pytest_collectreport"]
) )
def test_invalid_report_types(self, testdir, pytestconfig, hook_name): def test_invalid_report_types(
self, testdir: Testdir, pytestconfig: Config, hook_name: str
) -> None:
testdir.makepyfile( testdir.makepyfile(
""" """
def test_a(): pass def test_a(): pass

View File

@ -1,18 +1,21 @@
import os import os
from io import StringIO from io import StringIO
from typing import List
import _pytest._code import _pytest._code
import pytest import pytest
from _pytest.pytester import Testdir
from _pytest.resultlog import pytest_configure from _pytest.resultlog import pytest_configure
from _pytest.resultlog import pytest_unconfigure from _pytest.resultlog import pytest_unconfigure
from _pytest.resultlog import ResultLog from _pytest.resultlog import ResultLog
from _pytest.resultlog import resultlog_key from _pytest.resultlog import resultlog_key
pytestmark = pytest.mark.filterwarnings("ignore:--result-log is deprecated") pytestmark = pytest.mark.filterwarnings("ignore:--result-log is deprecated")
def test_write_log_entry(): def test_write_log_entry() -> None:
reslog = ResultLog(None, None) reslog = ResultLog(None, None) # type: ignore[arg-type]
reslog.logfile = StringIO() reslog.logfile = StringIO()
reslog.write_log_entry("name", ".", "") reslog.write_log_entry("name", ".", "")
entry = reslog.logfile.getvalue() entry = reslog.logfile.getvalue()
@ -54,14 +57,14 @@ class TestWithFunctionIntegration:
# XXX (hpk) i think that the resultlog plugin should # XXX (hpk) i think that the resultlog plugin should
# provide a Parser object so that one can remain # provide a Parser object so that one can remain
# ignorant regarding formatting details. # ignorant regarding formatting details.
def getresultlog(self, testdir, arg): def getresultlog(self, testdir: Testdir, arg: str) -> List[str]:
resultlog = testdir.tmpdir.join("resultlog") resultlog = testdir.tmpdir.join("resultlog")
testdir.plugins.append("resultlog") testdir.plugins.append("resultlog")
args = ["--resultlog=%s" % resultlog] + [arg] args = ["--resultlog=%s" % resultlog] + [arg]
testdir.runpytest(*args) testdir.runpytest(*args)
return [x for x in resultlog.readlines(cr=0) if x] return [x for x in resultlog.readlines(cr=0) if x]
def test_collection_report(self, testdir): def test_collection_report(self, testdir: Testdir) -> None:
ok = testdir.makepyfile(test_collection_ok="") ok = testdir.makepyfile(test_collection_ok="")
fail = testdir.makepyfile(test_collection_fail="XXX") fail = testdir.makepyfile(test_collection_fail="XXX")
lines = self.getresultlog(testdir, ok) lines = self.getresultlog(testdir, ok)
@ -75,7 +78,7 @@ class TestWithFunctionIntegration:
assert x.startswith(" ") assert x.startswith(" ")
assert "XXX" in "".join(lines[1:]) assert "XXX" in "".join(lines[1:])
def test_log_test_outcomes(self, testdir): def test_log_test_outcomes(self, testdir: Testdir) -> None:
mod = testdir.makepyfile( mod = testdir.makepyfile(
test_mod=""" test_mod="""
import pytest import pytest
@ -111,16 +114,17 @@ class TestWithFunctionIntegration:
assert len(lines) == 15 assert len(lines) == 15
@pytest.mark.parametrize("style", ("native", "long", "short")) @pytest.mark.parametrize("style", ("native", "long", "short"))
def test_internal_exception(self, style): def test_internal_exception(self, style) -> None:
# they are produced for example by a teardown failing # they are produced for example by a teardown failing
# at the end of the run or a failing hook invocation # at the end of the run or a failing hook invocation
try: try:
raise ValueError raise ValueError
except ValueError: except ValueError:
excinfo = _pytest._code.ExceptionInfo.from_current() excinfo = _pytest._code.ExceptionInfo.from_current()
reslog = ResultLog(None, StringIO()) file = StringIO()
reslog = ResultLog(None, file) # type: ignore[arg-type]
reslog.pytest_internalerror(excinfo.getrepr(style=style)) reslog.pytest_internalerror(excinfo.getrepr(style=style))
entry = reslog.logfile.getvalue() entry = file.getvalue()
entry_lines = entry.splitlines() entry_lines = entry.splitlines()
assert entry_lines[0].startswith("! ") assert entry_lines[0].startswith("! ")
@ -130,7 +134,7 @@ class TestWithFunctionIntegration:
assert "ValueError" in entry assert "ValueError" in entry
def test_generic(testdir, LineMatcher): def test_generic(testdir: Testdir, LineMatcher) -> None:
testdir.plugins.append("resultlog") testdir.plugins.append("resultlog")
testdir.makepyfile( testdir.makepyfile(
""" """
@ -162,7 +166,7 @@ def test_generic(testdir, LineMatcher):
) )
def test_makedir_for_resultlog(testdir, LineMatcher): def test_makedir_for_resultlog(testdir: Testdir, LineMatcher) -> None:
"""--resultlog should automatically create directories for the log file""" """--resultlog should automatically create directories for the log file"""
testdir.plugins.append("resultlog") testdir.plugins.append("resultlog")
testdir.makepyfile( testdir.makepyfile(
@ -177,7 +181,7 @@ def test_makedir_for_resultlog(testdir, LineMatcher):
LineMatcher(lines).fnmatch_lines([". *:test_pass"]) LineMatcher(lines).fnmatch_lines([". *:test_pass"])
def test_no_resultlog_on_workers(testdir): def test_no_resultlog_on_workers(testdir: Testdir) -> None:
config = testdir.parseconfig("-p", "resultlog", "--resultlog=resultlog") config = testdir.parseconfig("-p", "resultlog", "--resultlog=resultlog")
assert resultlog_key not in config._store assert resultlog_key not in config._store
@ -186,14 +190,14 @@ def test_no_resultlog_on_workers(testdir):
pytest_unconfigure(config) pytest_unconfigure(config)
assert resultlog_key not in config._store assert resultlog_key not in config._store
config.workerinput = {} config.workerinput = {} # type: ignore[attr-defined]
pytest_configure(config) pytest_configure(config)
assert resultlog_key not in config._store assert resultlog_key not in config._store
pytest_unconfigure(config) pytest_unconfigure(config)
assert resultlog_key not in config._store assert resultlog_key not in config._store
def test_unknown_teststatus(testdir): def test_unknown_teststatus(testdir: Testdir) -> None:
"""Ensure resultlog correctly handles unknown status from pytest_report_teststatus """Ensure resultlog correctly handles unknown status from pytest_report_teststatus
Inspired on pytest-rerunfailures. Inspired on pytest-rerunfailures.
@ -229,7 +233,7 @@ def test_unknown_teststatus(testdir):
assert lines[0] == "r test_unknown_teststatus.py::test" assert lines[0] == "r test_unknown_teststatus.py::test"
def test_failure_issue380(testdir): def test_failure_issue380(testdir: Testdir) -> None:
testdir.makeconftest( testdir.makeconftest(
""" """
import pytest import pytest