mirror of
https://github.com/esphome/esphome.git
synced 2026-10-11 13:54:12 +00:00
88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
"""Integration test for status_momentary_warning/error without a name."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from aioesphomeapi import BinarySensorState, EntityState
|
|
import pytest
|
|
|
|
from .state_utils import InitialStateHelper
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_momentary(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
loop = asyncio.get_running_loop()
|
|
warning_logged = loop.create_future()
|
|
error_logged = loop.create_future()
|
|
|
|
def on_log_line(line: str) -> None:
|
|
if (
|
|
"set Warning flag: probe warning reason" in line
|
|
and not warning_logged.done()
|
|
):
|
|
warning_logged.set_result(True)
|
|
if "set Error flag: probe error reason" in line and not error_logged.done():
|
|
error_logged.set_result(True)
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=on_log_line),
|
|
api_client_connected() as client,
|
|
):
|
|
entities, services = await client.list_entities_services()
|
|
svc = {s.name: s for s in services}
|
|
keys = {e.object_id: e.key for e in entities}
|
|
warning_key = keys["probe_warning"]
|
|
error_key = keys["probe_error"]
|
|
|
|
changes: asyncio.Queue[tuple[int, bool, float]] = asyncio.Queue()
|
|
|
|
def on_state(state: EntityState) -> None:
|
|
if isinstance(state, BinarySensorState):
|
|
changes.put_nowait((state.key, state.state, loop.time()))
|
|
|
|
initial_state_helper = InitialStateHelper(entities)
|
|
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
|
|
await initial_state_helper.wait_for_initial_states()
|
|
assert initial_state_helper.initial_states[warning_key].state is False
|
|
assert initial_state_helper.initial_states[error_key].state is False
|
|
|
|
async def next_change(key: int) -> tuple[bool, float]:
|
|
while True:
|
|
k, value, when = await asyncio.wait_for(changes.get(), timeout=5)
|
|
if k == key:
|
|
return value, when
|
|
|
|
# The flag sets at once and clears after the length
|
|
for service, key in (
|
|
("momentary_warning", warning_key),
|
|
("momentary_error", error_key),
|
|
):
|
|
start = loop.time()
|
|
await client.execute_service(svc[service], {"length": 300})
|
|
assert (await next_change(key))[0] is True
|
|
value, cleared = await next_change(key)
|
|
assert value is False
|
|
assert cleared - start >= 0.25
|
|
|
|
# A second call restarts the timeout instead of adding a second one
|
|
start = loop.time()
|
|
await client.execute_service(svc["momentary_warning"], {"length": 1000})
|
|
assert (await next_change(warning_key))[0] is True
|
|
await asyncio.sleep(0.6)
|
|
await client.execute_service(svc["momentary_warning"], {"length": 1000})
|
|
value, cleared = await next_change(warning_key)
|
|
assert value is False
|
|
assert cleared - start >= 1.4
|
|
|
|
# A message replaces "unspecified" in the log line
|
|
await client.execute_service(svc["momentary_warning_message"], {})
|
|
await asyncio.wait_for(warning_logged, timeout=5)
|
|
await client.execute_service(svc["momentary_error_message"], {})
|
|
await asyncio.wait_for(error_logged, timeout=5)
|