Fix typing in `reporting.py` and `cucumber_json.py`

I managed to remove all occurrences of `Any`, and use proper typed dicts instead
This commit is contained in:
Alessio Bogon 2024-12-01 16:34:23 +01:00
parent 4ccb683ecd
commit ad221becd1
No known key found for this signature in database
4 changed files with 110 additions and 29 deletions

View File

@ -6,11 +6,11 @@ import json
import math
import os
import time
from typing import TYPE_CHECKING, Any, Literal, TypedDict
from typing import TYPE_CHECKING, Literal, TypedDict
from typing_extensions import NotRequired
from .reporting import test_report_context_registry
from .reporting import FeatureDict, ScenarioReportDict, StepReportDict, test_report_context_registry
if TYPE_CHECKING:
from _pytest.config import Config
@ -19,6 +19,56 @@ if TYPE_CHECKING:
from _pytest.terminal import TerminalReporter
class ResultElementDict(TypedDict):
status: Literal["passed", "failed", "skipped"]
duration: int # in nanoseconds
error_message: NotRequired[str]
class TagElementDict(TypedDict):
name: str
line: int
class MatchElementDict(TypedDict):
location: str
class StepElementDict(TypedDict):
keyword: str
name: str
line: int
match: MatchElementDict
result: ResultElementDict
class ScenarioElementDict(TypedDict):
keyword: str
id: str
name: str
line: int
description: str
tags: list[TagElementDict]
type: Literal["scenario"]
steps: list[StepElementDict]
class FeatureElementDict(TypedDict):
keyword: str
uri: str
name: str
id: str
line: int
description: str
language: str
tags: list[TagElementDict]
elements: list[ScenarioElementDict]
class FeaturesDict(TypedDict):
features: dict[str, FeatureElementDict]
def add_options(parser: Parser) -> None:
"""Add pytest-bdd options."""
group = parser.getgroup("bdd", "Cucumber JSON")
@ -48,21 +98,15 @@ def unconfigure(config: Config) -> None:
config.pluginmanager.unregister(xml)
class Result(TypedDict):
status: Literal["passed", "failed", "skipped"]
duration: int # in nanoseconds
error_message: NotRequired[str]
class LogBDDCucumberJSON:
"""Logging plugin for cucumber like json output."""
def __init__(self, logfile: str) -> None:
logfile = os.path.expanduser(os.path.expandvars(logfile))
self.logfile = os.path.normpath(os.path.abspath(logfile))
self.features: dict[str, dict] = {}
self.features: dict[str, FeatureElementDict] = {}
def _get_result(self, step: dict[str, Any], report: TestReport, error_message: bool = False) -> Result:
def _get_result(self, step: StepReportDict, report: TestReport, error_message: bool = False) -> ResultElementDict:
"""Get scenario test run result.
:param step: `Step` step we get result for
@ -80,12 +124,12 @@ class LogBDDCucumberJSON:
status = "skipped"
else:
raise ValueError(f"Unknown test outcome {report.outcome}")
res: Result = {"status": status, "duration": int(math.floor((10**9) * step["duration"]))} # nanosec
res: ResultElementDict = {"status": status, "duration": int(math.floor((10**9) * step["duration"]))} # nanosec
if res_message is not None:
res["error_message"] = res_message
return res
def _serialize_tags(self, item: dict[str, Any]) -> list[dict[str, Any]]:
def _serialize_tags(self, item: FeatureDict | ScenarioReportDict) -> list[TagElementDict]:
"""Serialize item's tags.
:param item: json-serialized `Scenario` or `Feature`.
@ -110,7 +154,7 @@ class LogBDDCucumberJSON:
# skip if there isn't a result or scenario has no steps
return
def stepmap(step: dict[str, Any]) -> dict[str, Any]:
def stepmap(step: StepReportDict) -> StepElementDict:
error_message = False
if step["failed"] and not scenario.setdefault("failed", False):
scenario["failed"] = True

View File

@ -43,10 +43,10 @@ def configure(config: Config) -> None:
raise Exception("gherkin-terminal-reporter is not compatible with 'xdist' plugin.")
class GherkinTerminalReporter(TerminalReporter): # type: ignore
class GherkinTerminalReporter(TerminalReporter): # type: ignore[misc]
def __init__(self, config: Config) -> None:
super().__init__(config)
self.current_rule = None
self.current_rule: str | None = None
def pytest_runtest_logreport(self, report: TestReport) -> None:
rep = report

View File

@ -64,7 +64,7 @@ class Feature:
scenarios (OrderedDict[str, ScenarioTemplate]): A dictionary of scenarios in the feature.
filename (str): The absolute path of the feature file.
rel_filename (str): The relative path of the feature file.
name (Optional[str]): The name of the feature.
name (str): The name of the feature.
tags (set[str]): A set of tags associated with the feature.
background (Optional[Background]): The background steps for the feature, if any.
line_number (int): The line number where the feature starts in the file.
@ -76,7 +76,7 @@ class Feature:
rel_filename: str
language: str
keyword: str
name: str | None
name: str
tags: set[str]
background: Background | None
line_number: int

View File

@ -8,12 +8,12 @@ from __future__ import annotations
import time
from dataclasses import dataclass
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Callable, TypedDict
from weakref import WeakKeyDictionary
if TYPE_CHECKING:
from typing import Any, Callable
from typing_extensions import NotRequired
if TYPE_CHECKING:
from _pytest.fixtures import FixtureRequest
from _pytest.nodes import Item
from _pytest.reports import TestReport
@ -25,6 +25,44 @@ scenario_reports_registry: WeakKeyDictionary[Item, ScenarioReport] = WeakKeyDict
test_report_context_registry: WeakKeyDictionary[TestReport, ReportContext] = WeakKeyDictionary()
class FeatureDict(TypedDict):
keyword: str
name: str
filename: str
rel_filename: str
language: str
line_number: int
description: str
tags: list[str]
class RuleDict(TypedDict):
keyword: str
name: str
description: str
tags: list[str]
class StepReportDict(TypedDict):
name: str
type: str
keyword: str
line_number: int
failed: bool
duration: float
class ScenarioReportDict(TypedDict):
steps: list[StepReportDict]
keyword: str
name: str
line_number: int
tags: list[str]
feature: FeatureDict
rule: NotRequired[RuleDict]
failed: NotRequired[bool]
class StepReport:
"""Step execution report."""
@ -39,11 +77,10 @@ class StepReport:
self.step = step
self.started = time.perf_counter()
def serialize(self) -> dict[str, object]:
def serialize(self) -> StepReportDict:
"""Serialize the step execution report.
:return: Serialized step execution report.
:rtype: dict
"""
return {
"name": self.step.name,
@ -103,16 +140,15 @@ class ScenarioReport:
"""
self.step_reports.append(step_report)
def serialize(self) -> dict[str, object]:
def serialize(self) -> ScenarioReportDict:
"""Serialize scenario execution report in order to transfer reporting from nodes in the distributed mode.
:return: Serialized report.
:rtype: dict
"""
scenario = self.scenario
feature = scenario.feature
serialized = {
serialized: ScenarioReportDict = {
"steps": [step_report.serialize() for step_report in self.step_reports],
"keyword": scenario.keyword,
"name": scenario.name,
@ -131,12 +167,13 @@ class ScenarioReport:
}
if scenario.rule:
serialized["rule"] = {
rule_dict: RuleDict = {
"keyword": scenario.rule.keyword,
"name": scenario.rule.name,
"description": scenario.rule.description,
"tags": scenario.rule.tags,
"tags": sorted(scenario.rule.tags),
}
serialized["rule"] = rule_dict
return serialized
@ -154,7 +191,7 @@ class ScenarioReport:
@dataclass
class ReportContext:
scenario: dict[str, Any]
scenario: ScenarioReportDict
name: str
@ -191,7 +228,7 @@ def before_step(
feature: Feature,
scenario: Scenario,
step: Step,
step_func: Callable[..., Any],
step_func: Callable[..., object],
) -> None:
"""Store step start time."""
scenario_reports_registry[request.node].add_step_report(StepReport(step=step))