|
6 | 6 |
|
7 | 7 | import asyncio |
8 | 8 | import enum |
| 9 | +import re |
9 | 10 | from datetime import timedelta |
10 | 11 |
|
11 | 12 | import async_solipsism |
@@ -331,6 +332,18 @@ async def test_timer_construction_wrong_args() -> None: |
331 | 332 | loop=None, |
332 | 333 | ) |
333 | 334 |
|
| 335 | + with pytest.raises( |
| 336 | + ValueError, |
| 337 | + match=re.escape("`auto_start` must be `True` if `tick_at_start` is `True`"), |
| 338 | + ): |
| 339 | + _ = Timer( |
| 340 | + timedelta(seconds=5.0), |
| 341 | + SkipMissedAndResync(), |
| 342 | + auto_start=False, |
| 343 | + tick_at_start=True, |
| 344 | + loop=None, |
| 345 | + ) |
| 346 | + |
334 | 347 |
|
335 | 348 | async def test_timer_close_receiver() -> None: |
336 | 349 | """Test the autostart of a periodic timer.""" |
@@ -384,6 +397,46 @@ async def test_timer_autostart_with_delay() -> None: |
384 | 397 | assert event_loop.time() == pytest.approx(2.5) |
385 | 398 |
|
386 | 399 |
|
| 400 | +async def test_timer_autostart_with_tick_at_start() -> None: |
| 401 | + """Test the autostart of a periodic timer with a tick at start.""" |
| 402 | + event_loop = asyncio.get_running_loop() |
| 403 | + |
| 404 | + timer = Timer(timedelta(seconds=1.0), TriggerAllMissed(), tick_at_start=True) |
| 405 | + |
| 406 | + # The first tick should be at 0.0, without any delay. |
| 407 | + drift = await timer.receive() |
| 408 | + assert drift == pytest.approx(timedelta(seconds=0.0)) |
| 409 | + assert event_loop.time() == pytest.approx(0.0) |
| 410 | + |
| 411 | + # The next tick should be at 1.0 |
| 412 | + drift = await timer.receive() |
| 413 | + assert drift == pytest.approx(timedelta(seconds=0.0)) |
| 414 | + assert event_loop.time() == pytest.approx(1.0) |
| 415 | + |
| 416 | + |
| 417 | +async def test_timer_autostart_with_delay_and_tick_at_start() -> None: |
| 418 | + """Test the autostart of a periodic timer with a start delay and tick at start.""" |
| 419 | + event_loop = asyncio.get_running_loop() |
| 420 | + |
| 421 | + timer = Timer( |
| 422 | + timedelta(seconds=1.0), |
| 423 | + TriggerAllMissed(), |
| 424 | + tick_at_start=True, |
| 425 | + start_delay=timedelta(seconds=0.5), |
| 426 | + ) |
| 427 | + |
| 428 | + # The first tick should be at 0.5, as soon as the start delay is over. |
| 429 | + await asyncio.sleep(0.3) |
| 430 | + drift = await timer.receive() |
| 431 | + assert drift == pytest.approx(timedelta(seconds=0.0)) |
| 432 | + assert event_loop.time() == pytest.approx(0.5) |
| 433 | + |
| 434 | + # The next tick should be at 1.5 |
| 435 | + drift = await timer.receive() |
| 436 | + assert drift == pytest.approx(timedelta(seconds=0.0)) |
| 437 | + assert event_loop.time() == pytest.approx(1.5) |
| 438 | + |
| 439 | + |
387 | 440 | class _StartMethod(enum.Enum): |
388 | 441 | RESET = enum.auto() |
389 | 442 | RECEIVE = enum.auto() |
|
0 commit comments