Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add fail on repeat #72

Merged
merged 14 commits into from
Jan 13, 2024
Prev Previous commit
Next Next commit
update repeater
tsv1 committed Jan 9, 2024
commit 2cd360bdb742b94414ccb44076a29a0ae42f249f
2 changes: 1 addition & 1 deletion tests/core/scenario_runner/test_run_scenario.py
Original file line number Diff line number Diff line change
@@ -181,7 +181,7 @@ async def test_multiple_steps_failed(*, runner: MonotonicScenarioRunner, dispatc


@pytest.mark.parametrize("interrupt_exception", (KeyboardInterrupt, Interrupted))
async def test_step_interruped(interrupt_exception: Type[BaseException], *, dispatcher_: Mock):
async def test_step_interrupted(interrupt_exception: Type[BaseException], *, dispatcher_: Mock):
with given:
exception = interrupt_exception()
step1_, step2_ = Mock(side_effect=exception), Mock(return_value=None)
5 changes: 3 additions & 2 deletions vedro/plugins/repeater/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from ._repeater import Repeater, RepeaterPlugin, RepeaterPluginTriggered
from ._repeater import Repeater, RepeaterPlugin, RepeaterExecutionInterrupted
from ._scheduler import RepeaterScenarioScheduler

__all__ = ("Repeater", "RepeaterPlugin", "RepeaterScenarioScheduler", "RepeaterPluginTriggered",)
__all__ = ("Repeater", "RepeaterPlugin", "RepeaterScenarioScheduler",
"RepeaterExecutionInterrupted",)
39 changes: 24 additions & 15 deletions vedro/plugins/repeater/_repeater.py
Original file line number Diff line number Diff line change
@@ -17,10 +17,18 @@

from ._scheduler import RepeaterScenarioScheduler

__all__ = ("Repeater", "RepeaterPlugin", "RepeaterPluginTriggered",)
__all__ = ("Repeater", "RepeaterPlugin", "RepeaterExecutionInterrupted",)


class RepeaterPluginTriggered(Interrupted):
class RepeaterExecutionInterrupted(Interrupted):
"""
Exception raised when the execution of scenario repetition is interrupted.

This exception is used within the RepeaterPlugin to signal an early termination
of the scenario repetition process. It is typically raised when the fail-fast
condition is met, i.e., if a scenario fails and the --fail-fast-on-repeat option
is enabled, indicating that further repetitions of the scenario should be stopped.
"""
pass


@@ -76,13 +84,13 @@ def on_arg_parsed(self, event: ArgParsedEvent) -> None:
if self._repeats_delay < 0.0:
raise ValueError("--repeats-delay must be >= 0.0")

if self._repeats_delay > 0.0 and self._repeats <= 1:
if (self._repeats_delay > 0.0) and (self._repeats <= 1):
raise ValueError("--repeats-delay must be used with --repeats > 1")

if self._fail_fast and self._repeats <= 1:
if self._fail_fast and (self._repeats <= 1):
raise ValueError("--fail-fast-on-repeat must be used with --repeats > 1")

if self._repeats <= 1:
if not self.is_repeating_enabled():
return

assert self._global_config is not None # for type checking
@@ -92,15 +100,12 @@ def on_startup(self, event: StartupEvent) -> None:
self._scheduler = event.scheduler

def on_scenario_execute(self, event: Union[ScenarioRunEvent, ScenarioSkippedEvent]) -> None:
if not self._fail_fast:
return

if self._fail_fast and self._failed_count >= 1:
raise RepeaterPluginTriggered("Stop repeating scenarios after the first failure")
if self._fail_fast and (self._failed_count >= 1):
raise RepeaterExecutionInterrupted("Stop repeating scenarios after the first failure")

async def on_scenario_end(self,
event: Union[ScenarioPassedEvent, ScenarioFailedEvent]) -> None:
if self._repeats <= 1:
if not self.is_repeating_enabled():
return
assert isinstance(self._scheduler, RepeaterScenarioScheduler) # for type checking

@@ -114,17 +119,21 @@ async def on_scenario_end(self,
if self._repeat_count < self._repeats:
self._scheduler.schedule(scenario)

if self._repeats_delay > 0.0 and self._repeat_count < self._repeats:
if (self._repeats_delay > 0.0) and (self._repeat_count < self._repeats):
await self._sleep(self._repeats_delay)

if event.scenario_result.is_failed():
self._failed_count = 1

def on_cleanup(self, event: CleanupEvent) -> None:
if self._repeats <= 1:
if not self.is_repeating_enabled():
return
message = self._get_summary_message()
event.report.add_summary(message)
if not self._fail_fast or (self._failed_count == 0):
message = self._get_summary_message()
event.report.add_summary(message)

def is_repeating_enabled(self) -> bool:
return self._repeats > 1

def _get_summary_message(self) -> str:
message = f"repeated x{self._repeats}"