mirror of
https://github.com/esphome/esphome.git
synced 2026-10-10 13:33:06 +00:00
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""A background scheduler insertion must interrupt the loop's current sleep.
|
|
|
|
Covers both set_timeout and defer: a zero-delay insert from another thread
|
|
takes the separate defer queue on multi-threaded builds.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
import re
|
|
|
|
from aioesphomeapi import ButtonInfo
|
|
import pytest
|
|
|
|
from .state_utils import require_entity
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduler_background_wake(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
external_components_path = str(
|
|
Path(__file__).parent / "fixtures" / "external_components"
|
|
)
|
|
yaml_config = yaml_config.replace(
|
|
"EXTERNAL_COMPONENT_PATH", external_components_path
|
|
)
|
|
|
|
loop = asyncio.get_running_loop()
|
|
results: dict[int, asyncio.Future[tuple[int, int]]] = {
|
|
100: loop.create_future(),
|
|
0: loop.create_future(),
|
|
}
|
|
|
|
def on_log_line(line: str) -> None:
|
|
match = re.search(
|
|
r"SCHEDULER_WAKE_RESULT delay=(\d+) elapsed=(\d+) loop_delta=(-?\d+)",
|
|
line,
|
|
)
|
|
if match is None:
|
|
return
|
|
result = results[int(match.group(1))]
|
|
if not result.done():
|
|
result.set_result((int(match.group(2)), int(match.group(3))))
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=on_log_line),
|
|
api_client_connected() as client,
|
|
):
|
|
device_info = await client.device_info()
|
|
assert device_info is not None
|
|
assert device_info.name == "scheduler-background-wake"
|
|
entities, _ = await client.list_entities_services()
|
|
for object_id, delay in (
|
|
("start_scheduler_timeout", 100),
|
|
("start_scheduler_defer", 0),
|
|
):
|
|
start_button = require_entity(
|
|
entities,
|
|
object_id,
|
|
ButtonInfo,
|
|
description=f"{object_id} button",
|
|
)
|
|
client.button_command(start_button.key)
|
|
|
|
try:
|
|
elapsed, loop_delta = await asyncio.wait_for(
|
|
results[delay], timeout=10.0
|
|
)
|
|
except TimeoutError:
|
|
pytest.fail(f"background insert with delay={delay} did not fire")
|
|
|
|
assert elapsed < 1000, (
|
|
f"background insert with delay={delay} should interrupt the "
|
|
f"five-second sleep; it fired after {elapsed}ms"
|
|
)
|
|
assert loop_delta == 0, (
|
|
f"scheduler-only wake must not run component loops; observed "
|
|
f"loop_delta={loop_delta} for delay={delay}"
|
|
)
|