Skip to content

Commit aa0bac0

Browse files
committed
Add test for reset() while a timer is being waited on
This test needs to be an integration test because async-solipsism would make any sleep return immediately, so we wouldn't have a way to do the reset while the timer is being waited on. Signed-off-by: Leandro Lucarella <[email protected]>
1 parent ac01d09 commit aa0bac0

File tree

1 file changed

+31
-0
lines changed

1 file changed

+31
-0
lines changed

tests/test_timer_integration.py

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
"""Integration tests for the timer."""
2+
3+
4+
import asyncio
5+
from datetime import timedelta
6+
7+
import async_solipsism
8+
import pytest
9+
10+
from frequenz.channels.timer import Timer
11+
12+
13+
@pytest.mark.integration
14+
async def test_timer_timeout_reset(
15+
event_loop: async_solipsism.EventLoop, # pylint: disable=redefined-outer-name
16+
) -> None:
17+
"""Test that the receiving is properly adjusted after a reset."""
18+
19+
async def timer_wait(timer: Timer) -> None:
20+
await timer.receive()
21+
22+
async with asyncio.timeout(2.0):
23+
async with asyncio.TaskGroup() as task_group:
24+
timer = Timer.timeout(timedelta(seconds=1.0))
25+
start_time = event_loop.time()
26+
task_group.create_task(timer_wait(timer))
27+
await asyncio.sleep(0.5)
28+
timer.reset()
29+
30+
run_time = event_loop.time() - start_time
31+
assert run_time >= 1.5

0 commit comments

Comments
 (0)