diff --git a/README.md b/README.md index 0f37c0d..7b3bf39 100644 --- a/README.md +++ b/README.md @@ -20,13 +20,13 @@ $ pip3 install vedro-gitlab-reporter ```python # ./vedro.cfg.py import vedro -import vedro_gitlab_reporter as v +import vedro_gitlab_reporter as gitlab_reporter class Config(vedro.Config): class Plugins(vedro.Config.Plugins): - class GitlabReporter(v.GitlabReporter): + class GitlabReporter(gitlab_reporter.GitlabReporter): enabled = True ``` diff --git a/requirements-dev.txt b/requirements-dev.txt index b2f98ee..67670d0 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,10 +1,10 @@ baby-steps==1.2.1 bump2version==1.0.1 codecov==2.1.12 -coverage==6.3.2 -flake8==4.0.1 +coverage==6.4.4 +flake8==5.0.4 isort==5.10.1 -mypy==0.942 -pytest==7.1.2 -pytest-asyncio==0.18.3 +mypy==0.971 +pytest==7.1.3 +pytest-asyncio==0.19.0 pytest-cov==3.0.0 diff --git a/requirements.txt b/requirements.txt index 0389f25..6bf7279 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1 @@ -vedro>=1.5,<2.0 +vedro>=1.7,<2.0 diff --git a/setup.cfg b/setup.cfg index 348690b..193bd81 100644 --- a/setup.cfg +++ b/setup.cfg @@ -41,6 +41,6 @@ python_files = test_*.py python_classes = python_functions = test_* markers = only -asyncio_mode = strict +asyncio_mode = auto filterwarnings = ignore:.*verbose.*deprecated.*collapsable.*:DeprecationWarning diff --git a/setup.py b/setup.py index 32bb1aa..d4272da 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ def find_dev_required(): python_requires=">=3.7", url="https://github.com/nikitanovosibirsk/vedro-gitlab-reporter", license="Apache-2.0", - packages=find_packages(exclude=("tests",)), + packages=find_packages(exclude=["tests", "tests.*"]), package_data={"vedro_gitlab_reporter": ["py.typed"]}, install_requires=find_required(), tests_require=find_dev_required(), diff --git a/tests/_utils.py b/tests/_utils.py new file mode 100644 index 0000000..da5ab30 --- /dev/null +++ b/tests/_utils.py @@ -0,0 +1,123 @@ +import sys +from argparse import ArgumentParser, Namespace +from contextlib import contextmanager +from pathlib import Path +from time import monotonic_ns +from types import TracebackType +from typing import Optional, Union, cast +from unittest.mock import Mock, patch +from uuid import uuid4 + +import pytest +from vedro import Config, Scenario +from vedro.core import ( + AggregatedResult, + Dispatcher, + ExcInfo, + ScenarioResult, + StepResult, + VirtualScenario, + VirtualStep, +) +from vedro.events import ArgParsedEvent, ArgParseEvent, ConfigLoadedEvent, ScenarioRunEvent +from vedro.plugins.director import Director, DirectorPlugin +from vedro.plugins.director.rich import RichPrinter + +from vedro_gitlab_reporter import GitlabCollapsableMode, GitlabReporter, GitlabReporterPlugin + + +@pytest.fixture() +def dispatcher() -> Dispatcher: + return Dispatcher() + + +@pytest.fixture() +def printer_() -> Mock: + return Mock(RichPrinter) + + +@pytest.fixture() +def director(dispatcher: Dispatcher) -> DirectorPlugin: + director = DirectorPlugin(Director) + director.subscribe(dispatcher) + return director + + +@pytest.fixture() +def gitlab_reporter(dispatcher: Dispatcher, + director: DirectorPlugin, printer_: Mock) -> GitlabReporterPlugin: + reporter = GitlabReporterPlugin(GitlabReporter, printer_factory=lambda: printer_) + reporter.subscribe(dispatcher) + return reporter + + +async def fire_arg_parsed_event(dispatcher: Dispatcher, *, + collapsable_mode: Union[GitlabCollapsableMode, None] = None, + tb_show_internal_calls: bool = + GitlabReporter.tb_show_internal_calls, + tb_show_locals: bool = + GitlabReporter.tb_show_locals) -> None: + await dispatcher.fire(ConfigLoadedEvent(Path(), Config)) + + arg_parse_event = ArgParseEvent(ArgumentParser()) + await dispatcher.fire(arg_parse_event) + + namespace = Namespace(gitlab_collapsable=collapsable_mode, + gitlab_tb_show_internal_calls=tb_show_internal_calls, + gitlab_tb_show_locals=tb_show_locals) + arg_parsed_event = ArgParsedEvent(namespace) + await dispatcher.fire(arg_parsed_event) + + +async def fire_scenario_run_event(dispatcher: Dispatcher, + scenario_result: Optional[ScenarioResult] = None + ) -> ScenarioResult: + if scenario_result is None: + scenario_result = make_scenario_result() + scenario_run_event = ScenarioRunEvent(scenario_result) + await dispatcher.fire(scenario_run_event) + return scenario_result + + +def make_vstep(name: Optional[str] = None) -> VirtualStep: + def step(): + pass + step.__name__ = name or f"step_{monotonic_ns()}" + return VirtualStep(step) + + +def make_vscenario() -> VirtualScenario: + class _Scenario(Scenario): + __file__ = Path(f"scenario_{monotonic_ns()}.py").absolute() + + return VirtualScenario(_Scenario, steps=[]) + + +def make_step_result(vstep: Optional[VirtualStep] = None) -> StepResult: + return StepResult(vstep or make_vstep()) + + +def make_scenario_result(vscenario: Optional[VirtualScenario] = None) -> ScenarioResult: + return ScenarioResult(vscenario or make_vscenario()) + + +def make_aggregated_result(scenario_result: Optional[ScenarioResult] = None) -> AggregatedResult: + if scenario_result is None: + scenario_result = make_scenario_result() + return AggregatedResult.from_existing(scenario_result, [scenario_result]) + + +def make_exc_info(exc_val: Exception) -> ExcInfo: + try: + raise exc_val + except type(exc_val): + *_, traceback = sys.exc_info() + return ExcInfo(type(exc_val), exc_val, cast(TracebackType, traceback)) + + +@contextmanager +def patch_uuid(uuid: Optional[str] = None): + if uuid is None: + uuid = str(uuid4()) + with patch("uuid.uuid4", Mock(return_value=uuid)): + yield uuid diff --git a/tests/test_collapsable_mode.py b/tests/test_collapsable_mode.py new file mode 100644 index 0000000..afe90f2 --- /dev/null +++ b/tests/test_collapsable_mode.py @@ -0,0 +1,17 @@ +import pytest +from baby_steps import then, when + +from vedro_gitlab_reporter import GitlabCollapsableMode + + +@pytest.mark.parametrize(("mode", "text"), [ + (GitlabCollapsableMode.STEPS, "steps"), + (GitlabCollapsableMode.VARS, "vars"), + (GitlabCollapsableMode.SCOPE, "scope"), +]) +def test_collapsable_mode(mode: GitlabCollapsableMode, text: str): + with when: + res = str(mode) + + with then: + assert res == text diff --git a/tests/test_gitlab_collapsable.py b/tests/test_gitlab_collapsable.py new file mode 100644 index 0000000..0bb7788 --- /dev/null +++ b/tests/test_gitlab_collapsable.py @@ -0,0 +1,137 @@ +from unittest.mock import Mock, call + +import pytest +from baby_steps import given, then, when +from vedro.core import Dispatcher, ScenarioStatus, StepStatus +from vedro.events import ScenarioReportedEvent, StepFailedEvent + +from vedro_gitlab_reporter import GitlabCollapsableMode + +from ._utils import ( + director, + dispatcher, + fire_arg_parsed_event, + fire_scenario_run_event, + gitlab_reporter, + make_aggregated_result, + make_step_result, + patch_uuid, + printer_, +) + +__all__ = ("dispatcher", "director", "gitlab_reporter", "printer_") # fixtures + + +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_collapsable_steps(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher, collapsable_mode=GitlabCollapsableMode.STEPS) + scenario_result = await fire_scenario_run_event(dispatcher) + scenario_result.set_scope({"key": "val"}) + + step_result = make_step_result().mark_failed().set_started_at(1.0).set_ended_at(3.0) + await dispatcher.fire(StepFailedEvent(step_result)) + scenario_result.add_step_result(step_result) + + aggregated_result = make_aggregated_result(scenario_result.mark_failed()) + event = ScenarioReportedEvent(aggregated_result) + + printer_.reset_mock() + printer_.pretty_format = lambda self: "'val'" + + with when, patch_uuid() as uuid: + await dispatcher.fire(event) + + with then: + section_start, section_end = int(step_result.started_at), int(step_result.ended_at) + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.FAILED, + elapsed=aggregated_result.elapsed, + prefix=" "), + + call.console.file.write( + f"\x1b[0Ksection_start:{section_start}:{uuid}[collapsed=true]\r\x1b[0K"), + call.print_step_name(step_result.step_name, + StepStatus.FAILED, + elapsed=step_result.elapsed, + prefix=" " * 3), + call.print_scope_key("key", indent=5, line_break=True), + call.print_scope_val("'val'"), + call.console.file.write(f"\x1b[0Ksection_end:{section_end}:{uuid}\r\x1b[0K"), + ] + + +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_collapsable_vars(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher, collapsable_mode=GitlabCollapsableMode.VARS) + scenario_result = await fire_scenario_run_event(dispatcher) + scenario_result.set_scope({"key": "val"}) + + step_result = make_step_result().mark_failed().set_started_at(1.0).set_ended_at(3.0) + await dispatcher.fire(StepFailedEvent(step_result)) + scenario_result.add_step_result(step_result) + + aggregated_result = make_aggregated_result(scenario_result.mark_failed()) + event = ScenarioReportedEvent(aggregated_result) + + printer_.reset_mock() + printer_.pretty_format = lambda self: "'val'" + + with when, patch_uuid() as uuid: + await dispatcher.fire(event) + + with then: + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.FAILED, + elapsed=aggregated_result.elapsed, + prefix=" "), + + call.print_step_name(step_result.step_name, + StepStatus.FAILED, + elapsed=step_result.elapsed, + prefix=" " * 3), + + call.console.file.write(f"\x1b[0Ksection_start:0:{uuid}[collapsed=true]\r\x1b[0K"), + call.print_scope_key("key", indent=5, line_break=True), + call.print_scope_val("'val'"), + call.console.file.write(f"\x1b[0Ksection_end:0:{uuid}\r\x1b[0K"), + ] + + +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_collapsable_scope(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher, collapsable_mode=GitlabCollapsableMode.SCOPE) + scenario_result = await fire_scenario_run_event(dispatcher) + scenario_result.set_scope({"key": "val"}) + + step_result = make_step_result().mark_failed().set_started_at(1.0).set_ended_at(3.0) + await dispatcher.fire(StepFailedEvent(step_result)) + scenario_result.add_step_result(step_result) + + aggregated_result = make_aggregated_result(scenario_result.mark_failed()) + event = ScenarioReportedEvent(aggregated_result) + + printer_.reset_mock() + + with when, patch_uuid() as uuid: + await dispatcher.fire(event) + + with then: + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.FAILED, + elapsed=aggregated_result.elapsed, + prefix=" "), + call.print_step_name(step_result.step_name, + StepStatus.FAILED, + elapsed=step_result.elapsed, + prefix=" " * 3), + + call.console.file.write(f"\x1b[0Ksection_start:0:{uuid}[collapsed=true]\r\x1b[0K"), + call.print_scope(scenario_result.scope), + call.console.file.write(f"\x1b[0Ksection_end:0:{uuid}\r\x1b[0K"), + ] diff --git a/tests/test_gitlab_reporter.py b/tests/test_gitlab_reporter.py index 66bc2a7..3b245bd 100644 --- a/tests/test_gitlab_reporter.py +++ b/tests/test_gitlab_reporter.py @@ -1,73 +1,65 @@ -from argparse import Namespace -from contextlib import contextmanager -from typing import Optional, Union -from unittest.mock import Mock, call, patch -from uuid import uuid4 +from unittest.mock import Mock, call import pytest from baby_steps import given, then, when -from rich.console import Console -from rich.style import Style -from vedro.core import Dispatcher -from vedro.events import ArgParsedEvent, ScenarioFailedEvent, ScenarioRunEvent, StepFailedEvent -from vedro.plugins.director import DirectorPlugin, Reporter -from vedro.plugins.director.rich.test_utils import ( - chose_reporter, - console_, +from vedro.core import AggregatedResult, Dispatcher +from vedro.core import MonotonicScenarioScheduler as ScenarioScheduler +from vedro.core import Report, ScenarioStatus +from vedro.events import CleanupEvent, ScenarioReportedEvent, ScenarioRunEvent, StartupEvent +from vedro.plugins.director import DirectorInitEvent, DirectorPlugin + +from vedro_gitlab_reporter import GitlabReporter, GitlabReporterPlugin + +from ._utils import ( director, dispatcher, + fire_arg_parsed_event, + gitlab_reporter, + make_aggregated_result, make_scenario_result, - make_step_result, + printer_, ) -from vedro_gitlab_reporter import GitlabCollapsableMode, GitlabReporter, GitlabReporterPlugin +__all__ = ("dispatcher", "director", "gitlab_reporter", "printer_") # fixtures -__all__ = ("dispatcher", "console_", "director", "chose_reporter",) +async def test_subscribe(*, dispatcher: Dispatcher): + with given: + director_ = Mock(DirectorPlugin) -@pytest.fixture() -def reporter(dispatcher: Dispatcher, console_: Console) -> GitlabReporterPlugin: - reporter = GitlabReporterPlugin(GitlabReporter, console_factory=lambda: console_) - reporter.subscribe(dispatcher) - return reporter + reporter = GitlabReporterPlugin(GitlabReporter) + reporter.subscribe(dispatcher) + with when: + await dispatcher.fire(DirectorInitEvent(director_)) -@contextmanager -def patch_uuid(uuid: Optional[str] = None): - if uuid is None: - uuid = str(uuid4()) - with patch("uuid.uuid4", Mock(return_value=uuid)): - yield uuid + with then: + assert director_.mock_calls == [ + call.register("gitlab", reporter) + ] -def make_parsed_args(*, - verbose: int = 0, - gitlab_collapsable: Union[GitlabCollapsableMode, None] = None) -> Namespace: - return Namespace( - verbose=verbose, - gitlab_collapsable=gitlab_collapsable, - show_timings=False, - show_paths=False, - tb_show_internal_calls=False, - tb_show_locals=False, - reruns=0, - ) +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_startup(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher) + scheduler = ScenarioScheduler([]) + event = StartupEvent(scheduler) -def test_gitlab_reporter(): with when: - reporter = GitlabReporterPlugin(GitlabReporter) + await dispatcher.fire(event) with then: - assert isinstance(reporter, Reporter) + assert printer_.mock_calls == [ + call.print_header() + ] -@pytest.mark.asyncio -async def test_reporter_scenario_run_event(*, dispatcher: Dispatcher, - director: DirectorPlugin, - reporter: GitlabReporterPlugin, console_: Mock): +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_run(*, dispatcher: Dispatcher, printer_: Mock): with given: - await chose_reporter(dispatcher, director, reporter) + await fire_arg_parsed_event(dispatcher) scenario_result = make_scenario_result() event = ScenarioRunEvent(scenario_result) @@ -76,103 +68,142 @@ async def test_reporter_scenario_run_event(*, dispatcher: Dispatcher, await dispatcher.fire(event) with then: - assert console_.mock_calls == [ - call.out(f"* {scenario_result.scenario.namespace}", style=Style.parse("bold")) + assert printer_.print_namespace.assert_called() is None + assert len(printer_.mock_calls) == 1 + + +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_run_same_namespace(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher) + + scenario_result1 = make_scenario_result() + await dispatcher.fire(ScenarioRunEvent(scenario_result1)) + printer_.reset_mock() + + scenario_result2 = make_scenario_result() + event = ScenarioRunEvent(scenario_result2) + + with when: + await dispatcher.fire(event) + + with then: + assert printer_.print_namespace.assert_not_called() is None + assert len(printer_.mock_calls) == 0 + + +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_passed(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher) + + scenario_result = make_scenario_result().mark_passed() + aggregated_result = make_aggregated_result(scenario_result) + event = ScenarioReportedEvent(aggregated_result) + + with when: + await dispatcher.fire(event) + + with then: + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.PASSED, + elapsed=aggregated_result.elapsed, + prefix=" ") ] -@pytest.mark.parametrize("args", [ - make_parsed_args(verbose=0), # backward compatibility - make_parsed_args(gitlab_collapsable=None), -]) -@pytest.mark.asyncio -async def test_reporter_scenario_failed_event_verbose0(args: Namespace, *, - dispatcher: Dispatcher, - director: DirectorPlugin, - reporter: GitlabReporterPlugin, - console_: Mock): +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_failed(*, dispatcher: Dispatcher, printer_: Mock): with given: - await chose_reporter(dispatcher, director, reporter) - await dispatcher.fire(ArgParsedEvent(args)) + await fire_arg_parsed_event(dispatcher) scenario_result = make_scenario_result().mark_failed() - event = ScenarioFailedEvent(scenario_result) + aggregated_result = make_aggregated_result(scenario_result) + event = ScenarioReportedEvent(aggregated_result) with when: await dispatcher.fire(event) with then: - assert console_.mock_calls == [ - call.out(f" ✗ {scenario_result.scenario.subject}", style=Style.parse("red")) + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.FAILED, + elapsed=aggregated_result.elapsed, + prefix=" ") ] -@pytest.mark.parametrize("args", [ - make_parsed_args(verbose=1), # backward compatibility - make_parsed_args(gitlab_collapsable=GitlabCollapsableMode.STEPS), -]) -@pytest.mark.asyncio -async def test_reporter_scenario_failed_event_verbose1(args: Namespace, *, - dispatcher: Dispatcher, - director: DirectorPlugin, - reporter: GitlabReporterPlugin, - console_: Mock): +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_passed_aggregated_result(*, dispatcher: Dispatcher, printer_: Mock): with given: - await chose_reporter(dispatcher, director, reporter) - await dispatcher.fire(ArgParsedEvent(args)) + await fire_arg_parsed_event(dispatcher) + + scenario_results = [ + make_scenario_result().mark_passed(), + make_scenario_result().mark_passed(), + ] - step_result = make_step_result().mark_failed().set_started_at(1.0).set_ended_at(3.0) - scenario_result = make_scenario_result(step_results=[step_result]).mark_failed() - event = ScenarioFailedEvent(scenario_result) + aggregated_result = AggregatedResult.from_existing(scenario_results[0], scenario_results) + event = ScenarioReportedEvent(aggregated_result) - with when, patch_uuid() as uuid: + with when: await dispatcher.fire(event) with then: - assert console_.mock_calls == [ - call.out(f" ✗ {scenario_result.scenario.subject}", style=Style.parse("red")), - call.file.write(f"\x1b[0Ksection_start:{int(step_result.started_at)}:{uuid}" - "[collapsed=true]\r\x1b[0K"), - call.out(f" ✗ {step_result.step_name}", style=Style.parse("red")), - call.file.write(f"\x1b[0Ksection_end:{int(step_result.ended_at)}:{uuid}\r\x1b[0K") + assert printer_.mock_calls == [ + call.print_scenario_subject(aggregated_result.scenario.subject, + ScenarioStatus.PASSED, + elapsed=None, + prefix=" "), + + call.print_scenario_subject(aggregated_result.scenario_results[0].scenario.subject, + ScenarioStatus.PASSED, + elapsed=scenario_results[0].elapsed, + prefix=" │\n ├─[1/2] "), + call.print_scenario_subject(aggregated_result.scenario_results[1].scenario.subject, + ScenarioStatus.PASSED, + elapsed=scenario_results[1].elapsed, + prefix=" │\n ├─[2/2] "), + + call.print_empty_line(), ] -@pytest.mark.parametrize("args", [ - make_parsed_args(verbose=2), # backward compatibility - make_parsed_args(gitlab_collapsable=GitlabCollapsableMode.VARS), -]) -@pytest.mark.asyncio -async def test_reporter_scenario_failed_event_verbose2(args: Namespace, *, - dispatcher: Dispatcher, - director: DirectorPlugin, - reporter: GitlabReporterPlugin, - console_: Mock): +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_scenario_unknown_status(*, dispatcher: Dispatcher, printer_: Mock): with given: - await chose_reporter(dispatcher, director, reporter) - await dispatcher.fire(ArgParsedEvent(args)) + await fire_arg_parsed_event(dispatcher) scenario_result = make_scenario_result() - await dispatcher.fire(ScenarioRunEvent(scenario_result)) - console_.reset_mock() + aggregated_result = make_aggregated_result(scenario_result) + event = ScenarioReportedEvent(aggregated_result) + + with when: + await dispatcher.fire(event) + + with then: + assert printer_.mock_calls == [] - scenario_result.set_scope({"key": "val"}) - step_result = make_step_result().mark_failed() - await dispatcher.fire(StepFailedEvent(step_result)) - scenario_result = scenario_result.mark_failed() - scenario_result.add_step_result(step_result) - event = ScenarioFailedEvent(scenario_result) +@pytest.mark.usefixtures(gitlab_reporter.__name__) +async def test_cleanup(*, dispatcher: Dispatcher, printer_: Mock): + with given: + await fire_arg_parsed_event(dispatcher) - with when, patch_uuid() as uuid: + report = Report() + event = CleanupEvent(report) + + with when: await dispatcher.fire(event) with then: - assert console_.mock_calls == [ - call.out(f" ✗ {scenario_result.scenario.subject}", style=Style.parse("red")), - call.out(f" ✗ {step_result.step_name}", style=Style.parse("red")), - call.file.write(f"\x1b[0Ksection_start:0:{uuid}[collapsed=true]\r\x1b[0K"), - call.out(" key: ", style=Style.parse("blue")), - call.out("\"val\""), - call.file.write(f"\x1b[0Ksection_end:0:{uuid}\r\x1b[0K") + assert printer_.mock_calls == [ + call.print_empty_line(), + call.print_report_summary(report.summary), + call.print_report_stats(total=report.total, + passed=report.passed, + failed=report.failed, + skipped=report.skipped, + elapsed=report.elapsed) ] diff --git a/vedro_gitlab_reporter/__init__.py b/vedro_gitlab_reporter/__init__.py index 0ee3736..7ae9593 100644 --- a/vedro_gitlab_reporter/__init__.py +++ b/vedro_gitlab_reporter/__init__.py @@ -1,4 +1,5 @@ -from ._gitlab_reporter import GitlabCollapsableMode, GitlabReporter, GitlabReporterPlugin +from ._collapsable_mode import GitlabCollapsableMode +from ._gitlab_reporter import GitlabReporter, GitlabReporterPlugin __version__ = "1.0.1" __all__ = ("GitlabReporter", "GitlabReporterPlugin", "GitlabCollapsableMode",) diff --git a/vedro_gitlab_reporter/_collapsable_mode.py b/vedro_gitlab_reporter/_collapsable_mode.py new file mode 100644 index 0000000..01f6375 --- /dev/null +++ b/vedro_gitlab_reporter/_collapsable_mode.py @@ -0,0 +1,12 @@ +from enum import Enum + +__all__ = ("GitlabCollapsableMode",) + + +class GitlabCollapsableMode(Enum): + STEPS = "steps" + VARS = "vars" + SCOPE = "scope" + + def __str__(self) -> str: + return self.value diff --git a/vedro_gitlab_reporter/_gitlab_reporter.py b/vedro_gitlab_reporter/_gitlab_reporter.py index 4762c30..ea8ab15 100644 --- a/vedro_gitlab_reporter/_gitlab_reporter.py +++ b/vedro_gitlab_reporter/_gitlab_reporter.py @@ -1,170 +1,234 @@ +import operator import uuid -import warnings -from enum import Enum -from typing import Any, Dict, Set, Type, Union +from functools import reduce +from typing import Callable, Dict, Set, Type, Union -from rich.style import Style -from vedro.core import Dispatcher, ScenarioResult +from vedro.core import Dispatcher, PluginConfig, ScenarioResult from vedro.events import ( ArgParsedEvent, ArgParseEvent, + CleanupEvent, + ScenarioReportedEvent, ScenarioRunEvent, + StartupEvent, StepFailedEvent, StepPassedEvent, ) -from vedro.plugins.director import DirectorInitEvent, RichReporter, RichReporterPlugin +from vedro.plugins.director import DirectorInitEvent, Reporter +from vedro.plugins.director.rich import RichPrinter -__all__ = ("GitlabReporter", "GitlabReporterPlugin", "GitlabCollapsableMode",) +from ._collapsable_mode import GitlabCollapsableMode +__all__ = ("GitlabReporter", "GitlabReporterPlugin",) -class GitlabCollapsableMode(Enum): - STEPS = "steps" - VARS = "vars" - SCOPE = "scope" - def __str__(self) -> str: - return self.value +class GitlabReporterPlugin(Reporter): + def __init__(self, config: Type["GitlabReporter"], *, + printer_factory: Callable[[], RichPrinter] = RichPrinter) -> None: + super().__init__(config) + self._printer = printer_factory() + self._tb_show_internal_calls = config.tb_show_internal_calls + self._tb_show_locals = config.tb_show_locals + self._tb_max_frames = config.tb_max_frames + self._collapsable_mode: Union[GitlabCollapsableMode, None] = None -class GitlabReporterPlugin(RichReporterPlugin): - def __init__(self, config: Type["GitlabReporter"], **kwargs: Any) -> None: - super().__init__(config, **kwargs) + self._namespace: Union[str, None] = None self._scenario_result: Union[ScenarioResult, None] = None self._scenario_steps: Dict[str, Set[str]] = {} - self._prev_step_name: Union[str, None] = None - self._prev_scope: Set[str] = set() - self._collapsable_mode: Union[GitlabCollapsableMode, None] = None def subscribe(self, dispatcher: Dispatcher) -> None: - self._dispatcher = dispatcher + super().subscribe(dispatcher) dispatcher.listen(DirectorInitEvent, lambda e: e.director.register("gitlab", self)) def on_chosen(self) -> None: assert isinstance(self._dispatcher, Dispatcher) - super().on_chosen() - self._dispatcher.listen(StepPassedEvent, self.on_step_end) \ - .listen(StepFailedEvent, self.on_step_end) + self._dispatcher.listen(ArgParseEvent, self.on_arg_parse) \ + .listen(ArgParsedEvent, self.on_arg_parsed) \ + .listen(StartupEvent, self.on_startup) \ + .listen(ScenarioRunEvent, self.on_scenario_run) \ + .listen(StepPassedEvent, self.on_step_end) \ + .listen(StepFailedEvent, self.on_step_end) \ + .listen(ScenarioReportedEvent, self.on_scenario_reported) \ + .listen(CleanupEvent, self.on_cleanup) def on_arg_parse(self, event: ArgParseEvent) -> None: - super().on_arg_parse(event) group = event.arg_parser.add_argument_group("GitLab Reporter") group.add_argument("--gitlab-collapsable", type=GitlabCollapsableMode, choices=[x for x in GitlabCollapsableMode], help="Choose collapsable mode") + group.add_argument("--gitlab-tb-show-internal-calls", + action="store_true", + default=self._tb_show_internal_calls, + help="Show internal calls in the traceback output") + group.add_argument("--gitlab-tb-show-locals", + action="store_true", + default=self._tb_show_locals, + help="Show local variables in the traceback output") def on_arg_parsed(self, event: ArgParsedEvent) -> None: - super().on_arg_parsed(event) - - if self._verbosity > 0: - if event.args.gitlab_collapsable: - raise ValueError("Use --gitlab-collapsable or --verbose, but not both") - warnings.warn("GitlabReporterPlugin: " - "argument --verbose is deprecated, use --gitlab-collapsable instead", - DeprecationWarning) - self._collapsable_mode = event.args.gitlab_collapsable - # backward compatibility - if self._verbosity == 1: - self._collapsable_mode = GitlabCollapsableMode.STEPS - elif self._verbosity == 2: - self._collapsable_mode = GitlabCollapsableMode.VARS - elif self._verbosity == 3: - self._collapsable_mode = GitlabCollapsableMode.SCOPE + self._tb_show_internal_calls = event.args.gitlab_tb_show_internal_calls + self._tb_show_locals = event.args.gitlab_tb_show_locals + + def on_startup(self, event: StartupEvent) -> None: + self._printer.print_header() def on_scenario_run(self, event: ScenarioRunEvent) -> None: - super().on_scenario_run(event) + namespace = event.scenario_result.scenario.namespace + if namespace != self._namespace: + self._namespace = namespace + self._printer.print_namespace(namespace) + self._scenario_result = event.scenario_result self._scenario_steps = {} - self._prev_step_name = None - self._prev_scope = set() def on_step_end(self, event: Union[StepPassedEvent, StepFailedEvent]) -> None: assert isinstance(self._scenario_result, ScenarioResult) - step_name = event.step_result.step_name - if self._scenario_result.scope: - step_scope = set(self._scenario_result.scope.keys()) - else: - step_scope = set() + step_scope: Set[str] = set(self._scenario_result.scope.keys()) + prev_scope: Set[str] = reduce(operator.or_, self._scenario_steps.values(), set()) + self._scenario_steps[event.step_result.step_name] = step_scope - prev_scope + + def _print_scenario_result(self, scenario_result: ScenarioResult, *, prefix: str = "") -> None: + if scenario_result.is_passed(): + self._print_scenario_passed(scenario_result, prefix=prefix) + elif scenario_result.is_failed(): + self._print_scenario_failed(scenario_result, prefix=prefix) - self._scenario_steps[step_name] = step_scope - self._prev_scope - self._prev_scope = step_scope - self._prev_step_name = step_name + def _print_scenario_passed(self, scenario_result: ScenarioResult, *, prefix: str = "") -> None: + self._printer.print_scenario_subject(scenario_result.scenario.subject, + scenario_result.status, + elapsed=scenario_result.elapsed, + prefix=prefix) - def _print_scenario_failed(self, scenario_result: ScenarioResult, *, indent: int = 0) -> None: - self._print_scenario_subject(scenario_result, self._show_timings) + def _print_scenario_failed(self, scenario_result: ScenarioResult, *, prefix: str = "") -> None: + self._printer.print_scenario_subject(scenario_result.scenario.subject, + scenario_result.status, + elapsed=scenario_result.elapsed, + prefix=prefix) if self._collapsable_mode == GitlabCollapsableMode.STEPS: - self._print_collapsable_steps(scenario_result, indent=4 + indent) + prefix = self._prefix_to_indent(prefix, indent=2) + self._print_collapsable_steps(scenario_result, prefix=prefix) self._print_exceptions(scenario_result) elif self._collapsable_mode == GitlabCollapsableMode.VARS: - self._print_steps_with_collapsable_scope(scenario_result, indent=4 + indent) + prefix = self._prefix_to_indent(prefix, indent=2) + self._print_steps_with_collapsable_vars(scenario_result, prefix=prefix) self._print_exceptions(scenario_result) elif self._collapsable_mode == GitlabCollapsableMode.SCOPE: - self._print_steps(scenario_result, indent=4 + indent) + prefix = self._prefix_to_indent(prefix, indent=2) + self._print_steps(scenario_result, prefix=prefix) self._print_exceptions(scenario_result) self._print_collapsable_scope(scenario_result) - def _print_section_start(self, name: str, started_at: int = 0, + def on_scenario_reported(self, event: ScenarioReportedEvent) -> None: + aggregated_result = event.aggregated_result + rescheduled = len(aggregated_result.scenario_results) + if rescheduled == 1: + self._print_scenario_result(aggregated_result, prefix=" ") + return + + self._printer.print_scenario_subject(aggregated_result.scenario.subject, + aggregated_result.status, elapsed=None, prefix=" ") + for index, scenario_result in enumerate(aggregated_result.scenario_results, start=1): + prefix = f" │\n ├─[{index}/{rescheduled}] " + self._print_scenario_result(scenario_result, prefix=prefix) + + self._printer.print_empty_line() + + def on_cleanup(self, event: CleanupEvent) -> None: + self._printer.print_empty_line() + self._printer.print_report_summary(event.report.summary) + self._printer.print_report_stats(total=event.report.total, + passed=event.report.passed, + failed=event.report.failed, + skipped=event.report.skipped, + elapsed=event.report.elapsed) + + def _print_section_start(self, name: str, started_at: int = 0, *, is_collapsed: bool = True) -> None: collapsed = "true" if is_collapsed else "false" output = f'\033[0Ksection_start:{started_at}:{name}[collapsed={collapsed}]\r\033[0K' - self._console.file.write(output) + self._printer.console.file.write(output) def _print_section_end(self, name: str, ended_at: int = 0) -> None: output = f'\033[0Ksection_end:{ended_at}:{name}\r\033[0K' - self._console.file.write(output) + self._printer.console.file.write(output) - def _print_steps(self, scenario_result: ScenarioResult, *, indent: int = 0) -> None: - for step_result in scenario_result.step_results: - self._print_step_name(step_result, indent=indent) + def _prefix_to_indent(self, prefix: str, indent: int = 0) -> str: + last_line = prefix.split("\n")[-1] + return (len(last_line) + indent) * " " - def _print_exceptions(self, scenario_result: ScenarioResult) -> None: + def _print_steps(self, scenario_result: ScenarioResult, *, prefix: str = "") -> None: for step_result in scenario_result.step_results: - if step_result.exc_info: - self._print_exception(step_result.exc_info.value, - step_result.exc_info.traceback) + self._printer.print_step_name(step_result.step_name, step_result.status, + elapsed=step_result.elapsed, prefix=prefix) def _print_collapsable_steps(self, scenario_result: ScenarioResult, *, - indent: int = 0) -> None: + prefix: str = "") -> None: for step_result in scenario_result.step_results: section_name = str(uuid.uuid4()) started_at = int(step_result.started_at) if step_result.started_at else 0 self._print_section_start(section_name, started_at) - self._print_step_name(step_result, indent=indent) + self._printer.print_step_name(step_result.step_name, step_result.status, + elapsed=step_result.elapsed, prefix=prefix) - for key, val in self._format_scope(scenario_result.scope): - if key in self._scenario_steps[step_result.step_name]: - self._console.out(f"{indent * ' '} {key}: ", style=Style(color="blue")) - self._console.out(val) + for key, val in scenario_result.scope.items(): + if key not in self._scenario_steps[step_result.step_name]: + continue + self._printer.print_scope_key(key, indent=len(prefix) + 2, line_break=True) + self._printer.print_scope_val(self._printer.pretty_format(val)) ended_at = int(step_result.ended_at) if step_result.ended_at else 0 self._print_section_end(section_name, ended_at) - def _print_steps_with_collapsable_scope(self, scenario_result: ScenarioResult, *, - indent: int = 0) -> None: + def _print_steps_with_collapsable_vars(self, scenario_result: ScenarioResult, *, + prefix: str = "") -> None: for step_result in scenario_result.step_results: - self._print_step_name(step_result, indent=indent) + self._printer.print_step_name(step_result.step_name, step_result.status, + elapsed=step_result.elapsed, prefix=prefix) - for key, val in self._format_scope(scenario_result.scope): - if key in self._scenario_steps[step_result.step_name]: - section_name = str(uuid.uuid4()) - self._print_section_start(section_name) - self._console.out(f"{indent * ' '} {key}: ", style=Style(color="blue")) - self._console.out(val) - self._print_section_end(section_name) + for key, val in scenario_result.scope.items(): + if key not in self._scenario_steps[step_result.step_name]: + continue + section_name = str(uuid.uuid4()) + self._print_section_start(section_name) + + self._printer.print_scope_key(key, indent=len(prefix) + 2, line_break=True) + self._printer.print_scope_val(self._printer.pretty_format(val)) + + self._print_section_end(section_name) + + def _print_exceptions(self, scenario_result: ScenarioResult) -> None: + for step_result in scenario_result.step_results: + if step_result.exc_info is None: + continue + self._printer.print_pretty_exception(step_result.exc_info, + max_frames=self._tb_max_frames, + show_locals=self._tb_show_locals, + show_internal_calls=self._tb_show_internal_calls) def _print_collapsable_scope(self, scenario_result: ScenarioResult) -> None: section_name = str(uuid.uuid4()) self._print_section_start(section_name) - self._print_scope(scenario_result.scope) + self._printer.print_scope(scenario_result.scope) self._print_section_end(section_name) -class GitlabReporter(RichReporter): +class GitlabReporter(PluginConfig): plugin = GitlabReporterPlugin + + # Show internal calls in the traceback output + tb_show_internal_calls: bool = False + + # Show local variables in the traceback output + tb_show_locals: bool = False + + # Max stack trace entries to show (min=4) + tb_max_frames: int = 8