1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
"""log machine-parseable test session result information to a plain text file."""
import os
from typing import IO
from typing import Union
from _pytest._code.code import ExceptionRepr
from _pytest.config import Config
from _pytest.config.argparsing import Parser
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
from _pytest.store import StoreKey
resultlog_key = StoreKey["ResultLog"]()
def pytest_addoption(parser: Parser) -> None:
group = parser.getgroup("terminal reporting", "resultlog plugin options")
group.addoption(
"--resultlog",
"--result-log",
action="store",
metavar="path",
default=None,
help="DEPRECATED path for machine-readable result log.",
)
def pytest_configure(config: Config) -> None:
resultlog = config.option.resultlog
# Prevent opening resultlog on worker nodes (xdist).
if resultlog and not hasattr(config, "workerinput"):
dirname = os.path.dirname(os.path.abspath(resultlog))
if not os.path.isdir(dirname):
os.makedirs(dirname)
logfile = open(resultlog, "w", 1) # line buffered
config._store[resultlog_key] = ResultLog(config, logfile)
config.pluginmanager.register(config._store[resultlog_key])
from _pytest.deprecated import RESULT_LOG
from _pytest.warnings import _issue_warning_captured
_issue_warning_captured(RESULT_LOG, config.hook, stacklevel=2)
def pytest_unconfigure(config: Config) -> None:
resultlog = config._store.get(resultlog_key, None)
if resultlog:
resultlog.logfile.close()
del config._store[resultlog_key]
config.pluginmanager.unregister(resultlog)
class ResultLog:
def __init__(self, config: Config, logfile: IO[str]) -> None:
self.config = config
self.logfile = logfile # preferably line buffered
def write_log_entry(self, testpath: str, lettercode: str, longrepr: str) -> None:
print("{} {}".format(lettercode, testpath), file=self.logfile)
for line in longrepr.splitlines():
print(" %s" % line, file=self.logfile)
def log_outcome(
self, report: Union[TestReport, CollectReport], lettercode: str, longrepr: str
) -> None:
testpath = getattr(report, "nodeid", None)
if testpath is None:
testpath = report.fspath
self.write_log_entry(testpath, lettercode, longrepr)
def pytest_runtest_logreport(self, report: TestReport) -> None:
if report.when != "call" and report.passed:
return
res = self.config.hook.pytest_report_teststatus(
report=report, config=self.config
)
code = res[1] # type: str
if code == "x":
longrepr = str(report.longrepr)
elif code == "X":
longrepr = ""
elif report.passed:
longrepr = ""
elif report.skipped:
assert isinstance(report.longrepr, tuple)
longrepr = str(report.longrepr[2])
else:
longrepr = str(report.longrepr)
self.log_outcome(report, code, longrepr)
def pytest_collectreport(self, report: CollectReport) -> None:
if not report.passed:
if report.failed:
code = "F"
longrepr = str(report.longrepr)
else:
assert report.skipped
code = "S"
longrepr = "%s:%d: %s" % report.longrepr # type: ignore
self.log_outcome(report, code, longrepr)
def pytest_internalerror(self, excrepr: ExceptionRepr) -> None:
if excrepr.reprcrash is not None:
path = excrepr.reprcrash.path
else:
path = "cwd:%s" % os.getcwd()
self.write_log_entry(path, "!", str(excrepr))
|
