update libs (#3)

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
 ```
 

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

requirements.txt

@@ -1 +1 @@
-vedro>=1.5,<2.0
+vedro>=1.7,<2.0

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

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(),

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

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

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"),
+        ]