Skip to content

Commit fca8a85

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 2f82b55 commit fca8a85

File tree

1 file changed

+34
-0
lines changed

1 file changed

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

0 commit comments

Comments
 (0)