mirror of
https://github.com/home-assistant/core.git
synced 2026-05-16 13:31:01 +01:00
J. Nick Koston
b6b5b1f788
This test relied on the event loop overhead. Change it to listen for the state to change instead
427 lines
14 KiB
Python
427 lines
14 KiB
Python
"""Test ESPHome binary sensors."""
|
|
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
from aioesphomeapi import (
|
|
APIClient,
|
|
BinarySensorInfo,
|
|
BinarySensorState,
|
|
EntityInfo,
|
|
EntityState,
|
|
SensorInfo,
|
|
SensorState,
|
|
UserService,
|
|
)
|
|
|
|
from homeassistant.const import (
|
|
ATTR_RESTORED,
|
|
EVENT_HOMEASSISTANT_STOP,
|
|
STATE_OFF,
|
|
STATE_ON,
|
|
STATE_UNAVAILABLE,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.helpers.event import (
|
|
EventStateChangedData,
|
|
async_track_state_change_event,
|
|
)
|
|
from homeassistant.helpers.typing import EventType
|
|
|
|
from .conftest import MockESPHomeDevice
|
|
|
|
|
|
async def test_entities_removed(
|
|
hass: HomeAssistant,
|
|
mock_client: APIClient,
|
|
hass_storage: dict[str, Any],
|
|
mock_esphome_device: Callable[
|
|
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
|
|
Awaitable[MockESPHomeDevice],
|
|
],
|
|
) -> None:
|
|
"""Test entities are removed when static info changes."""
|
|
ent_reg = er.async_get(hass)
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor_to_be_removed",
|
|
key=2,
|
|
name="my binary_sensor to be removed",
|
|
unique_id="mybinary_sensor_to_be_removed",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
BinarySensorState(key=2, state=True, missing_state=False),
|
|
]
|
|
user_service = []
|
|
mock_device = await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
)
|
|
entry = mock_device.entry
|
|
entry_id = entry.entry_id
|
|
storage_key = f"esphome.{entry_id}"
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
|
|
await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass_storage[storage_key]["data"]["binary_sensor"]) == 2
|
|
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.attributes[ATTR_RESTORED] is True
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is not None
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is not None
|
|
assert state.attributes[ATTR_RESTORED] is True
|
|
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
]
|
|
mock_device = await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
entry=entry,
|
|
)
|
|
assert mock_device.entry.entry_id == entry_id
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is None
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is None
|
|
await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert len(hass_storage[storage_key]["data"]["binary_sensor"]) == 1
|
|
|
|
|
|
async def test_entities_removed_after_reload(
|
|
hass: HomeAssistant,
|
|
mock_client: APIClient,
|
|
hass_storage: dict[str, Any],
|
|
mock_esphome_device: Callable[
|
|
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
|
|
Awaitable[MockESPHomeDevice],
|
|
],
|
|
) -> None:
|
|
"""Test entities and their registry entry are removed when static info changes after a reload."""
|
|
ent_reg = er.async_get(hass)
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor_to_be_removed",
|
|
key=2,
|
|
name="my binary_sensor to be removed",
|
|
unique_id="mybinary_sensor_to_be_removed",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
BinarySensorState(key=2, state=True, missing_state=False),
|
|
]
|
|
user_service = []
|
|
mock_device: MockESPHomeDevice = await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
)
|
|
entry = mock_device.entry
|
|
entry_id = entry.entry_id
|
|
storage_key = f"esphome.{entry_id}"
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is not None
|
|
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass_storage[storage_key]["data"]["binary_sensor"]) == 2
|
|
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.attributes[ATTR_RESTORED] is True
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is not None
|
|
assert state.attributes[ATTR_RESTORED] is True
|
|
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is not None
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass_storage[storage_key]["data"]["binary_sensor"]) == 2
|
|
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert ATTR_RESTORED not in state.attributes
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is not None
|
|
assert ATTR_RESTORED not in state.attributes
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is not None
|
|
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
]
|
|
mock_device.client.list_entities_services = AsyncMock(
|
|
return_value=(entity_info, user_service)
|
|
)
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
on_future = hass.loop.create_future()
|
|
|
|
@callback
|
|
def _async_wait_for_on(event: EventType[EventStateChangedData]) -> None:
|
|
if event.data["new_state"].state == STATE_ON:
|
|
on_future.set_result(None)
|
|
|
|
async_track_state_change_event(
|
|
hass, ["binary_sensor.test_mybinary_sensor"], _async_wait_for_on
|
|
)
|
|
await hass.async_block_till_done()
|
|
async with asyncio.timeout(2):
|
|
await on_future
|
|
|
|
assert mock_device.entry.entry_id == entry_id
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert state is None
|
|
|
|
await hass.async_block_till_done()
|
|
|
|
reg_entry = ent_reg.async_get("binary_sensor.test_mybinary_sensor_to_be_removed")
|
|
assert reg_entry is None
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert len(hass_storage[storage_key]["data"]["binary_sensor"]) == 1
|
|
|
|
|
|
async def test_entity_info_object_ids(
|
|
hass: HomeAssistant,
|
|
mock_client: APIClient,
|
|
mock_esphome_device: Callable[
|
|
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
|
|
Awaitable[MockESPHomeDevice],
|
|
],
|
|
) -> None:
|
|
"""Test how object ids affect entity id."""
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="object_id_is_used",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
)
|
|
]
|
|
states = []
|
|
user_service = []
|
|
await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
)
|
|
state = hass.states.get("binary_sensor.test_object_id_is_used")
|
|
assert state is not None
|
|
|
|
|
|
async def test_deep_sleep_device(
|
|
hass: HomeAssistant,
|
|
mock_client: APIClient,
|
|
hass_storage: dict[str, Any],
|
|
mock_esphome_device: Callable[
|
|
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
|
|
Awaitable[MockESPHomeDevice],
|
|
],
|
|
) -> None:
|
|
"""Test a deep sleep device."""
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
SensorInfo(
|
|
object_id="my_sensor",
|
|
key=3,
|
|
name="my sensor",
|
|
unique_id="my_sensor",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
BinarySensorState(key=2, state=True, missing_state=False),
|
|
SensorState(key=3, state=123.0, missing_state=False),
|
|
]
|
|
user_service = []
|
|
mock_device = await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
device_info={"has_deep_sleep": True},
|
|
)
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == "123"
|
|
|
|
await mock_device.mock_disconnect(False)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
await mock_device.mock_connect()
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == "123"
|
|
|
|
await mock_device.mock_disconnect(True)
|
|
await hass.async_block_till_done()
|
|
await mock_device.mock_connect()
|
|
await hass.async_block_till_done()
|
|
mock_device.set_state(BinarySensorState(key=1, state=False, missing_state=False))
|
|
mock_device.set_state(SensorState(key=3, state=56, missing_state=False))
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_OFF
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == "56"
|
|
|
|
await mock_device.mock_disconnect(True)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_OFF
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == "56"
|
|
|
|
await mock_device.mock_connect()
|
|
await hass.async_block_till_done()
|
|
await mock_device.mock_disconnect(False)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
state = hass.states.get("sensor.test_my_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
await mock_device.mock_connect()
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
await hass.async_block_till_done()
|
|
# Verify we do not dispatch any more state updates or
|
|
# availability updates after the stop event is fired
|
|
state = hass.states.get("binary_sensor.test_mybinary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|
|
|
|
|
|
async def test_esphome_device_without_friendly_name(
|
|
hass: HomeAssistant,
|
|
mock_client: APIClient,
|
|
hass_storage: dict[str, Any],
|
|
mock_esphome_device: Callable[
|
|
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
|
|
Awaitable[MockESPHomeDevice],
|
|
],
|
|
) -> None:
|
|
"""Test a device without friendly_name set."""
|
|
entity_info = [
|
|
BinarySensorInfo(
|
|
object_id="mybinary_sensor",
|
|
key=1,
|
|
name="my binary_sensor",
|
|
unique_id="my_binary_sensor",
|
|
),
|
|
]
|
|
states = [
|
|
BinarySensorState(key=1, state=True, missing_state=False),
|
|
BinarySensorState(key=2, state=True, missing_state=False),
|
|
]
|
|
user_service = []
|
|
await mock_esphome_device(
|
|
mock_client=mock_client,
|
|
entity_info=entity_info,
|
|
user_service=user_service,
|
|
states=states,
|
|
device_info={"friendly_name": None},
|
|
)
|
|
state = hass.states.get("binary_sensor.my_binary_sensor")
|
|
assert state is not None
|
|
assert state.state == STATE_ON
|