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 13, 2024
commit 8a02f5ae0c59dc3756abe6c1a7e771a246d07355
74 changes: 11 additions & 63 deletions tests/plugins/repeater/test_repeater.py
Original file line number Diff line number Diff line change
@@ -17,7 +17,6 @@
dispatcher,
fire_arg_parsed_event,
fire_failed_event,
fire_passed_event,
fire_startup_event,
make_scenario_result,
repeater,
@@ -28,7 +27,7 @@
__all__ = ("dispatcher", "repeater", "scheduler_", "sleep_") # fixtures


@pytest.mark.parametrize("repeats", [2, 3])
@pytest.mark.parametrize("repeats", [1, 2, 3])
@pytest.mark.parametrize("get_event", [
lambda scn_result: ScenarioPassedEvent(scn_result.mark_passed()),
lambda scn_result: ScenarioFailedEvent(scn_result.mark_failed()),
@@ -47,58 +46,7 @@ async def test_repeat(repeats: int, get_event: Callable[[ScenarioResult], Event]
await dispatcher.fire(scenario_event)

with then:
assert scheduler_.mock_calls == [call.schedule(scenario_result.scenario)]
assert sleep_.mock_calls == []


@pytest.mark.parametrize("get_event", [
lambda scn_result: ScenarioPassedEvent(scn_result.mark_passed()),
lambda scn_result: ScenarioFailedEvent(scn_result.mark_failed()),
])
@pytest.mark.usefixtures(repeater.__name__)
async def test_repeat2_fired_twice(get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher, scheduler_: Mock, sleep_: AsyncMock):
with given:
await fire_arg_parsed_event(dispatcher, repeats=2)
await fire_startup_event(dispatcher, scheduler_)

scenario_result1 = await fire_passed_event(dispatcher)
scenario_result2 = make_scenario_result(scenario_result1.scenario)

scenario_event = get_event(scenario_result2)
scheduler_.reset_mock()

with when:
await dispatcher.fire(scenario_event)

with then:
assert scheduler_.mock_calls == []
assert sleep_.mock_calls == []


@pytest.mark.parametrize("repeats", [3, 4])
@pytest.mark.parametrize("get_event", [
lambda scn_result: ScenarioPassedEvent(scn_result.mark_passed()),
lambda scn_result: ScenarioFailedEvent(scn_result.mark_failed()),
])
@pytest.mark.usefixtures(repeater.__name__)
async def test_repeat3_fired_twice(repeats: int, get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher, scheduler_: Mock, sleep_: AsyncMock):
with given:
await fire_arg_parsed_event(dispatcher, repeats=repeats)
await fire_startup_event(dispatcher, scheduler_)

scenario_result1 = await fire_passed_event(dispatcher)
scenario_result2 = make_scenario_result(scenario_result1.scenario)

scenario_event = get_event(scenario_result2)
scheduler_.reset_mock()

with when:
await dispatcher.fire(scenario_event)

with then:
assert scheduler_.mock_calls == [call.schedule(scenario_result2.scenario)]
assert scheduler_.mock_calls == [call.schedule(scenario_result.scenario)] * (repeats - 1)
assert sleep_.mock_calls == []


@@ -126,8 +74,8 @@ async def test_dont_repeat_skipped(repeats: int, *,
(3, 1.0)
])
@pytest.mark.usefixtures(repeater.__name__)
async def test_repeat_with_delay(repeats: int, repeats_delay: float, *,
dispatcher: Dispatcher, scheduler_: Mock, sleep_: AsyncMock):
async def _test_repeat_with_delay(repeats: int, repeats_delay: float, *,
dispatcher: Dispatcher, scheduler_: Mock, sleep_: AsyncMock):
with given:
await fire_arg_parsed_event(dispatcher, repeats=repeats, repeats_delay=repeats_delay)
await fire_startup_event(dispatcher, scheduler_)
@@ -148,9 +96,9 @@ async def test_repeat_with_delay(repeats: int, repeats_delay: float, *,
lambda scn_result: ScenarioFailedEvent(scn_result.mark_failed()),
])
@pytest.mark.usefixtures(repeater.__name__)
async def test_repeat2_with_delay_fired_twice(get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher,
scheduler_: Mock, sleep_: AsyncMock):
async def _test_repeat2_with_delay_fired_twice(get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher,
scheduler_: Mock, sleep_: AsyncMock):
with given:
await fire_arg_parsed_event(dispatcher, repeats=2, repeats_delay=0.1)
await fire_startup_event(dispatcher, scheduler_)
@@ -176,10 +124,10 @@ async def test_repeat2_with_delay_fired_twice(get_event: Callable[[ScenarioResul
lambda scn_result: ScenarioFailedEvent(scn_result.mark_failed()),
])
@pytest.mark.usefixtures(repeater.__name__)
async def test_repeat3_with_delay_fired_twice(repeats: int,
get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher,
scheduler_: Mock, sleep_: AsyncMock):
async def _test_repeat3_with_delay_fired_twice(repeats: int,
get_event: Callable[[ScenarioResult], Event], *,
dispatcher: Dispatcher,
scheduler_: Mock, sleep_: AsyncMock):
with given:
repeats_delay = 0.1
await fire_arg_parsed_event(dispatcher, repeats=repeats, repeats_delay=repeats_delay)
2 changes: 1 addition & 1 deletion vedro/plugins/repeater/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from ._repeater import Repeater, RepeaterPlugin, RepeaterExecutionInterrupted
from ._repeater import Repeater, RepeaterExecutionInterrupted, RepeaterPlugin
from ._scheduler import RepeaterScenarioScheduler

__all__ = ("Repeater", "RepeaterPlugin", "RepeaterScenarioScheduler",
48 changes: 24 additions & 24 deletions vedro/plugins/repeater/_repeater.py
Original file line number Diff line number Diff line change
@@ -46,7 +46,6 @@ def __init__(self, config: Type["Repeater"], *, sleep: SleepType = asyncio.sleep
self._global_config: Union[ConfigType, None] = None
self._scheduler: Union[ScenarioScheduler, None] = None
self._repeat_scenario_id: Union[str, None] = None
self._repeat_count: int = 0
self._failed_count: int = 0

def subscribe(self, dispatcher: Dispatcher) -> None:
@@ -90,49 +89,50 @@ def on_arg_parsed(self, event: ArgParsedEvent) -> None:
if self._fail_fast and (self._repeats <= 1):
raise ValueError("--fail-fast-on-repeat must be used with --repeats > 1")

if not self.is_repeating_enabled():
return

assert self._global_config is not None # for type checking
self._global_config.Registry.ScenarioScheduler.register(self._scheduler_factory, self)
if self._is_repeating_enabled():
assert self._global_config is not None # for type checking
self._global_config.Registry.ScenarioScheduler.register(self._scheduler_factory, self)

def on_startup(self, event: StartupEvent) -> None:
self._scheduler = event.scheduler

def on_scenario_execute(self, event: Union[ScenarioRunEvent, ScenarioSkippedEvent]) -> None:
if self._fail_fast and (self._failed_count >= 1):
async def on_scenario_execute(self,
event: Union[ScenarioRunEvent, ScenarioSkippedEvent]) -> None:
if not self._is_repeating_enabled():
return

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

scenario = event.scenario_result.scenario
if (self._repeat_scenario_id == scenario.unique_id) and (self._repeats_delay > 0.0):
await self._sleep(self._repeats_delay)

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

scenario = event.scenario_result.scenario
if self._repeat_scenario_id != scenario.unique_id:
if scenario.unique_id != self._repeat_scenario_id:
self._repeat_scenario_id = scenario.unique_id
self._repeat_count = 1
self._scheduler.schedule(scenario)
else:
self._repeat_count += 1
if self._repeat_count < self._repeats:
self._failed_count = 1 if event.scenario_result.is_failed() else 0
for _ in range(self._repeats - 1):
self._scheduler.schedule(scenario)

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
else:
if event.scenario_result.is_failed():
self._failed_count += 1

def on_cleanup(self, event: CleanupEvent) -> None:
if not self.is_repeating_enabled():
if not self._is_repeating_enabled():
return
if not self._fail_fast or (self._failed_count == 0):

if not event.report.interrupted:
message = self._get_summary_message()
event.report.add_summary(message)

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

def _get_summary_message(self) -> str: