diff --git a/src/pytest_bdd/cucumber_json.py b/src/pytest_bdd/cucumber_json.py index 5d016ec2..a8a4cf49 100644 --- a/src/pytest_bdd/cucumber_json.py +++ b/src/pytest_bdd/cucumber_json.py @@ -6,11 +6,11 @@ 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 @@ 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 @@ def _get_result(self, step: dict[str, Any], report: TestReport, error_message: b 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 @@ def pytest_runtest_logreport(self, report: TestReport) -> None: # 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 diff --git a/src/pytest_bdd/gherkin_terminal_reporter.py b/src/pytest_bdd/gherkin_terminal_reporter.py index e92808cd..264aea2d 100644 --- a/src/pytest_bdd/gherkin_terminal_reporter.py +++ b/src/pytest_bdd/gherkin_terminal_reporter.py @@ -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 diff --git a/src/pytest_bdd/parser.py b/src/pytest_bdd/parser.py index 4da0d9c1..6bb15e47 100644 --- a/src/pytest_bdd/parser.py +++ b/src/pytest_bdd/parser.py @@ -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 diff --git a/src/pytest_bdd/reporting.py b/src/pytest_bdd/reporting.py index 2706a28e..fe047b4f 100644 --- a/src/pytest_bdd/reporting.py +++ b/src/pytest_bdd/reporting.py @@ -8,12 +8,12 @@ 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 @@ 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 @@ def __init__(self, step: Step) -> None: 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 @@ def add_step_report(self, step_report: StepReport) -> None: """ 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 @@ def serialize(self) -> dict[str, object]: } 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 @@ def fail(self) -> None: @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))