forked from pytest-dev/pytest-bdd
/
reporting.py
182 lines (146 loc) · 5.56 KB
/
reporting.py
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
"""Reporting functionality.
Collection of the scenario execution statuses, timing and other information
that enriches the pytest test reporting.
"""
from __future__ import annotations
import time
from typing import Any, Callable
import pytest
from attr import Factory, attrib, attrs
from pytest_bdd.model import Feature, Scenario, Step
from pytest_bdd.typing.pytest import CallInfo, FixtureRequest, Item
class StepReport:
"""StepHandler execution report."""
failed = False
stopped = None
def __init__(self, step: Step) -> None:
"""StepHandler report constructor.
:param StepHandler step: StepHandler.
"""
self.step = step
self.started = time.perf_counter()
def serialize(self) -> dict[str, Any]:
"""Serialize the step execution report.
:return: Serialized step execution report.
:rtype: dict
"""
return {
"name": self.step.name,
"type": self.step.prefix,
"keyword": self.step.keyword,
"line_number": self.step.line_number,
"failed": self.failed,
"duration": self.duration,
}
def finalize(self, failed: bool) -> None:
"""Stop collecting information and finalize the report.
:param bool failed: Whether the step execution is failed.
"""
self.stopped = time.perf_counter()
self.failed = failed
@property
def duration(self) -> float:
"""StepHandler execution duration.
:return: StepHandler execution duration.
:rtype: float
"""
if self.stopped is None:
return 0
return self.stopped - self.started
@attrs
class ScenarioReport:
"""Scenario execution report."""
feature: Feature = attrib()
scenario: Scenario = attrib()
step_reports: list[StepReport] = attrib(default=Factory(list))
@property
def current_step_report(self) -> StepReport:
"""Get current step report.
:return: Last or current step report.
:rtype: pytest_bdd.reporting.StepReport
"""
return self.step_reports[-1]
def add_step_report(self, step_report: StepReport) -> None:
"""Add new step report.
:param step_report: New current step report.
:type step_report: pytest_bdd.reporting.StepReport
"""
self.step_reports.append(step_report)
def serialize(self) -> dict[str, Any]:
"""Serialize scenario execution report in order to transfer reporting from nodes in the distributed mode.
:return: Serialized report.
:rtype: dict
"""
pickle = self.scenario
feature = self.feature
return {
"steps": [step_report.serialize() for step_report in self.step_reports],
"name": pickle.name,
"line_number": pickle.line_number,
"tags": sorted(set(pickle.tag_names).difference(feature.tag_names)),
"feature": {
"name": feature.name,
"filename": feature.filename,
"rel_filename": feature.rel_filename,
"line_number": feature.line_number,
"description": feature.description,
"tags": feature.tag_names,
},
}
def fail(self) -> None:
"""Stop collecting information and finalize the report as failed."""
self.current_step_report.finalize(failed=True)
remaining_steps = self.scenario.steps[len(self.step_reports) :]
# Fail the rest of the steps and make reports.
for step in remaining_steps:
report = StepReport(step=step)
report.finalize(failed=True)
self.add_step_report(report)
class ScenarioReporterPlugin:
def __init__(self):
self.current_report = None
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(self, item: Item, call: CallInfo):
outcome = yield
if call.when != "setup":
rep = outcome.get_result()
"""Store item in the report object."""
scenario_report: ScenarioReport = self.current_report
if scenario_report is not None:
rep.scenario = scenario_report.serialize()
rep.item = {"name": item.name}
@pytest.hookimpl(tryfirst=True)
def pytest_bdd_before_scenario(self, request: FixtureRequest, feature: Feature, scenario: Scenario) -> None:
"""Create scenario report for the item."""
self.current_report = ScenarioReport(feature=feature, scenario=scenario) # type: ignore[call-arg]
@pytest.hookimpl(tryfirst=True)
def pytest_bdd_step_error(
self,
request: FixtureRequest,
feature: Feature,
scenario: Scenario,
step: Step,
step_func: Callable,
step_func_args: dict,
exception: Exception,
) -> None:
"""Finalize the step report as failed."""
self.current_report.fail()
@pytest.hookimpl(tryfirst=True)
def pytest_bdd_before_step(
self, request: FixtureRequest, feature: Feature, scenario: Scenario, step: Step, step_func: Callable
) -> None:
"""Store step start time."""
self.current_report.add_step_report(StepReport(step=step))
@pytest.hookimpl(tryfirst=True)
def pytest_bdd_after_step(
self,
request: FixtureRequest,
feature: Feature,
scenario: Scenario,
step: Step,
step_func: Callable,
step_func_args: dict,
) -> None:
"""Finalize the step report as successful."""
self.current_report.current_step_report.finalize(failed=False)