mirror of
https://github.com/esphome/esphome.git
synced 2026-06-24 11:05:18 +00:00
68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
"""Test that Scheduler::set_timer_common_ coerces interval=0 to 1ms.
|
|
|
|
Regression test for the scheduler busy-loop when interval=0 was passed
|
|
literally. Without the coercion, Scheduler::call() would spin forever
|
|
because the item's next_execution == now_64 after re-scheduling, failing
|
|
the loop's `> now_64` break condition. The device would fail to yield
|
|
back to the main loop and trigger a WDT reset.
|
|
|
|
With the coercion, interval=0 becomes interval=1 and the scheduler
|
|
fires at ~1kHz (bounded by the loop), the main loop continues to run,
|
|
and the device stays responsive to API calls.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduler_interval_zero_coerced(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
"""interval=0ms must be coerced to 1ms and not starve the main loop."""
|
|
loop = asyncio.get_running_loop()
|
|
reached_50: asyncio.Future[None] = loop.create_future()
|
|
coerce_warning: asyncio.Future[None] = loop.create_future()
|
|
|
|
def on_log_line(line: str) -> None:
|
|
if "ZERO_INTERVAL_50_FIRES_REACHED" in line and not reached_50.done():
|
|
reached_50.set_result(None)
|
|
if "would spin main loop" in line and not coerce_warning.done():
|
|
coerce_warning.set_result(None)
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=on_log_line),
|
|
api_client_connected() as client,
|
|
):
|
|
# The API-client connection itself is evidence that the main loop
|
|
# is not starved — if set_interval(0) were spinning we could not
|
|
# get here at all.
|
|
device_info = await client.device_info()
|
|
assert device_info is not None
|
|
assert device_info.name == "sched-interval-zero"
|
|
|
|
# Coerce warning must fire at registration
|
|
try:
|
|
await asyncio.wait_for(coerce_warning, timeout=5.0)
|
|
except TimeoutError:
|
|
pytest.fail("Expected coerce warning 'would spin main loop' not seen")
|
|
|
|
# The coerced 1ms interval should fire 50 times quickly — this
|
|
# confirms the callback actually runs (not just registered) and the
|
|
# scheduler yields back to the main loop each time.
|
|
try:
|
|
await asyncio.wait_for(reached_50, timeout=5.0)
|
|
except TimeoutError:
|
|
pytest.fail(
|
|
"Coerced interval=0→1ms did not reach 50 fires within 5s, "
|
|
"which would indicate either the coercion failed or the "
|
|
"main loop is still being starved."
|
|
)
|