mirror of
https://github.com/esphome/esphome.git
synced 2026-10-07 11:26:39 +00:00
214 lines
8.5 KiB
Python
214 lines
8.5 KiB
Python
"""Test the homeassistant select, text and button platforms against a fake Home Assistant."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from aioesphomeapi import APIClient, HomeassistantServiceCall, UserService
|
|
import pytest
|
|
|
|
from .log_utils import LineWaiter
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
SUBSCRIPTIONS = {
|
|
("select.mode", "options"),
|
|
("select.mode", None),
|
|
("input_select.colour", "options"),
|
|
("input_select.colour", None),
|
|
("text.note", None),
|
|
("text.note", "min"),
|
|
("text.note", "max"),
|
|
("text.note", "mode"),
|
|
("input_text.name", None),
|
|
("input_text.name", "min"),
|
|
("input_text.name", "max"),
|
|
("input_text.name", "mode"),
|
|
}
|
|
|
|
# Each is rejected by the options parser, leaving the previous options in place
|
|
MALFORMED_OPTIONS = (
|
|
"None",
|
|
"",
|
|
"[",
|
|
"['a'",
|
|
"['a',",
|
|
"['a' 'b']",
|
|
"['a'] x",
|
|
"[1]",
|
|
"['a\\q']",
|
|
"['a\\",
|
|
"['\\x4']",
|
|
"['\\x0']",
|
|
"['\\x00']",
|
|
"['\\U00110000']",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_homeassistant_entities(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
"""Options, state and actions of the homeassistant select, text and button platforms."""
|
|
loop = asyncio.get_running_loop()
|
|
waiter = LineWaiter()
|
|
subscribed: set[tuple[str, str | None]] = set()
|
|
all_subscribed = loop.create_future()
|
|
calls: asyncio.Queue[HomeassistantServiceCall] = asyncio.Queue()
|
|
|
|
def on_state_sub(entity_id: str, attribute: str | None) -> None:
|
|
subscribed.add((entity_id, attribute or None))
|
|
if not all_subscribed.done() and subscribed >= SUBSCRIPTIONS:
|
|
all_subscribed.set_result(None)
|
|
|
|
async def next_call() -> HomeassistantServiceCall:
|
|
return await asyncio.wait_for(calls.get(), timeout=5.0)
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=waiter.callback),
|
|
api_client_connected() as client,
|
|
):
|
|
entities, services = await client.list_entities_services()
|
|
keys = {entity.name: entity.key for entity in entities}
|
|
dump_select = next(s for s in services if s.name == "dump_select")
|
|
client.subscribe_service_calls(calls.put_nowait)
|
|
client.subscribe_home_assistant_states(on_state_sub)
|
|
try:
|
|
await asyncio.wait_for(all_subscribed, timeout=5.0)
|
|
except TimeoutError:
|
|
pytest.fail(f"never subscribed: {SUBSCRIPTIONS - subscribed}")
|
|
|
|
await _check_select(client, waiter, keys, dump_select, next_call)
|
|
await _check_text(client, waiter, keys, next_call)
|
|
|
|
client.button_command(keys["HA Button"])
|
|
call = await next_call()
|
|
assert call.service == "button.press"
|
|
assert call.data == {"entity_id": "button.doorbell"}
|
|
client.button_command(keys["HA Input Button"])
|
|
call = await next_call()
|
|
assert call.service == "input_button.press"
|
|
assert call.data == {"entity_id": "input_button.reset"}
|
|
|
|
assert calls.empty()
|
|
|
|
|
|
async def _check_select(
|
|
client: APIClient,
|
|
waiter: LineWaiter,
|
|
keys: dict[str, int],
|
|
dump_select: UserService,
|
|
next_call: Callable[[], Awaitable[HomeassistantServiceCall]],
|
|
) -> None:
|
|
async def dump(tag: str) -> str:
|
|
await client.execute_service(dump_select, {"tag": tag})
|
|
return await waiter.wait_for(f"dump {tag}:", timeout=5.0)
|
|
|
|
# Built with str() exactly as Home Assistant does, so every escape it emits is covered
|
|
options = ["Low", "It's", "café\t\x7f\\", "\u200b\U000e0001\r\n'\""]
|
|
client.send_home_assistant_state("select.mode", "options", str(options))
|
|
await waiter.wait_for("'select.mode': Got 4 options", timeout=5.0)
|
|
|
|
client.send_home_assistant_state("select.mode", "", "It's")
|
|
await waiter.wait_for("select value 'It's' at 1", timeout=5.0)
|
|
assert "has_state=1 index=1 option='It's'" in await dump("first")
|
|
|
|
# Decoded options round trip through select.select_option unchanged
|
|
for option in options[2:]:
|
|
client.select_command(keys["HA Select"], option)
|
|
call = await next_call()
|
|
assert call.service == "select.select_option"
|
|
assert call.data == {"entity_id": "select.mode", "option": option}
|
|
|
|
# The active option keeps its state when it moves; double quoted options parse too
|
|
client.send_home_assistant_state("select.mode", "options", "[\"It's\", 'L\\\"ow']")
|
|
await waiter.wait_for("'select.mode': Got 2 options", timeout=5.0)
|
|
assert "has_state=1 index=0 option='It's'" in await dump("moved")
|
|
client.select_command(keys["HA Select"], 'L"ow')
|
|
assert (await next_call()).data["option"] == 'L"ow'
|
|
|
|
# Dropping the active option clears the state
|
|
client.send_home_assistant_state(
|
|
"select.mode", "options", "[ 'Low' , 'Max' , 'High' ]"
|
|
)
|
|
await waiter.wait_for("'select.mode': Got 3 options", timeout=5.0)
|
|
assert "has_state=0" in await dump("dropped")
|
|
|
|
client.send_home_assistant_state("select.mode", "", "unavailable")
|
|
await waiter.wait_for("State 'unavailable' is not one of the options", timeout=5.0)
|
|
|
|
# Rejected lists keep the previous options
|
|
client.send_home_assistant_state("select.mode", "options", str(list("abcde")))
|
|
await waiter.wait_for("5 options exceed max_options (4)", timeout=5.0)
|
|
client.send_home_assistant_state("select.mode", "options", str(["x" * 50]))
|
|
await waiter.wait_for(
|
|
"Options need 51 bytes, more than options_buffer_size (40)", timeout=5.0
|
|
)
|
|
for malformed in MALFORMED_OPTIONS:
|
|
client.send_home_assistant_state("select.mode", "options", malformed)
|
|
client.send_home_assistant_state("select.mode", "", "Max")
|
|
await waiter.wait_for("select value 'Max' at 1", timeout=5.0)
|
|
assert sum("Can't parse options" in line for line in waiter.lines) == len(
|
|
MALFORMED_OPTIONS
|
|
)
|
|
|
|
# Repeating the active state does not publish again
|
|
client.send_home_assistant_state("select.mode", "", "Max")
|
|
client.send_home_assistant_state("select.mode", "options", "[]")
|
|
await waiter.wait_for("'select.mode': Got 0 options", timeout=5.0)
|
|
assert sum("select value 'Max'" in line for line in waiter.lines) == 1
|
|
assert "has_state=0 index=-1" in await dump("empty")
|
|
|
|
client.send_home_assistant_state(
|
|
"input_select.colour", "options", "['Red', 'Green']"
|
|
)
|
|
await waiter.wait_for("'input_select.colour': Got 2 options", timeout=5.0)
|
|
client.select_command(keys["HA Input Select"], "Green")
|
|
call = await next_call()
|
|
assert call.service == "input_select.select_option"
|
|
assert call.data == {"entity_id": "input_select.colour", "option": "Green"}
|
|
|
|
|
|
async def _check_text(
|
|
client: APIClient,
|
|
waiter: LineWaiter,
|
|
keys: dict[str, int],
|
|
next_call: Callable[[], Awaitable[HomeassistantServiceCall]],
|
|
) -> None:
|
|
client.send_home_assistant_state("text.note", "min", "2")
|
|
client.send_home_assistant_state("text.note", "max", "10")
|
|
client.send_home_assistant_state("text.note", "mode", "password")
|
|
await waiter.wait_for("'text.note': Mode retrieved: password", timeout=5.0)
|
|
|
|
client.send_home_assistant_state("input_text.name", "min", "abc")
|
|
client.send_home_assistant_state("input_text.name", "max", "x")
|
|
client.send_home_assistant_state("input_text.name", "mode", "fancy")
|
|
await waiter.wait_for(
|
|
"'input_text.name': Unknown 'mode' value 'fancy'", timeout=5.0
|
|
)
|
|
await waiter.wait_for("Can't convert 'min' value 'abc'", timeout=5.0)
|
|
await waiter.wait_for("Can't convert 'max' value 'x'", timeout=5.0)
|
|
client.send_home_assistant_state("input_text.name", "mode", "text")
|
|
await waiter.wait_for("'input_text.name': Mode retrieved: text", timeout=5.0)
|
|
|
|
client.send_home_assistant_state("text.note", "", "hello")
|
|
client.send_home_assistant_state("text.note", "", "hello")
|
|
client.send_home_assistant_state("text.note", "", "there")
|
|
await waiter.wait_for("text value 'there'", timeout=5.0)
|
|
assert sum("text value 'hello'" in line for line in waiter.lines) == 1
|
|
|
|
# The limits from Home Assistant apply: too short and too long are refused
|
|
client.text_command(keys["HA Text"], "a")
|
|
client.text_command(keys["HA Text"], "abcdefghijk")
|
|
client.text_command(keys["HA Text"], "fine")
|
|
call = await next_call()
|
|
assert call.service == "text.set_value"
|
|
assert call.data == {"entity_id": "text.note", "value": "fine"}
|
|
|
|
client.text_command(keys["HA Input Text"], "Bob")
|
|
call = await next_call()
|
|
assert call.service == "input_text.set_value"
|
|
assert call.data == {"entity_id": "input_text.name", "value": "Bob"}
|