mirror of
https://github.com/home-assistant/core.git
synced 2026-10-06 14:29:21 -04:00
Fix Modbus entities holding up Home Assistant startup while their device is unreachable (#182621)
This commit is contained in:
@@ -1,9 +1,10 @@
|
||||
"""Base implementation for all modbus platforms."""
|
||||
|
||||
from abc import abstractmethod
|
||||
from collections.abc import Callable
|
||||
import asyncio
|
||||
from collections.abc import Callable, Coroutine
|
||||
import copy
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import datetime
|
||||
import struct
|
||||
from typing import Any, cast, override
|
||||
|
||||
@@ -21,7 +22,7 @@ from homeassistant.const import (
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity import Entity, ToggleEntity
|
||||
from homeassistant.helpers.event import async_call_later
|
||||
@@ -83,6 +84,8 @@ class ModbusBaseEntity(Entity):
|
||||
self._input_type = entry[CONF_INPUT_TYPE]
|
||||
self._scan_interval = int(entry[CONF_SCAN_INTERVAL])
|
||||
self._cancel_call: Callable[[], None] | None = None
|
||||
self._update_tasks: set[asyncio.Task[None]] = set()
|
||||
self._stopped = False
|
||||
self._attr_unique_id = entry.get(CONF_UNIQUE_ID)
|
||||
self._attr_name = entry[CONF_NAME]
|
||||
self._attr_device_class = entry.get(CONF_DEVICE_CLASS)
|
||||
@@ -108,40 +111,68 @@ class ModbusBaseEntity(Entity):
|
||||
self._cancel_call()
|
||||
await self._async_update()
|
||||
self.async_write_ha_state()
|
||||
if self._scan_interval > 0:
|
||||
self._cancel_call = async_call_later(
|
||||
self.hass,
|
||||
timedelta(seconds=self._scan_interval),
|
||||
self.async_local_update,
|
||||
if self._scan_interval > 0 and not self._stopped:
|
||||
# an overlapping update scheduled one already, keep a single timer
|
||||
if self._cancel_call:
|
||||
self._cancel_call()
|
||||
self._cancel_call = self._async_call_later(
|
||||
self._scan_interval, self.async_local_update
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Remove entity from hass."""
|
||||
self.async_disable()
|
||||
@callback
|
||||
def _async_call_later(
|
||||
self, delay: float, action: Callable[[], Coroutine[Any, Any, None]]
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Run an update after a delay, in the background.
|
||||
|
||||
Startup and shutdown wait for foreground tasks, and the first update
|
||||
waits for a device that may never connect.
|
||||
"""
|
||||
|
||||
@callback
|
||||
def _run(_now: datetime) -> None:
|
||||
if self._stopped:
|
||||
return
|
||||
task = self.hass.async_create_background_task(
|
||||
action(), f"modbus {self._attr_name} update"
|
||||
)
|
||||
# updates can overlap when an action runs one while a poll is due
|
||||
self._update_tasks.add(task)
|
||||
task.add_done_callback(self._update_tasks.discard)
|
||||
|
||||
return async_call_later(self.hass, delay, _run)
|
||||
|
||||
@callback
|
||||
def _async_cancel_updates(self) -> None:
|
||||
"""Cancel the scheduled update and the ones still running."""
|
||||
if self._cancel_call:
|
||||
self._cancel_call()
|
||||
self._cancel_call = None
|
||||
for task in self._update_tasks:
|
||||
task.cancel()
|
||||
|
||||
@callback
|
||||
def async_disable(self) -> None:
|
||||
"""Remote stop entity."""
|
||||
LOGGER.info(f"hold entity {self._attr_name}")
|
||||
if self._cancel_call:
|
||||
self._cancel_call()
|
||||
self._cancel_call = None
|
||||
# an update started by an action is not tracked, it must not start
|
||||
# polling again when it finishes
|
||||
self._stopped = True
|
||||
self._async_cancel_updates()
|
||||
self._attr_available = False
|
||||
|
||||
async def async_await_connection(self, _now: Any) -> None:
|
||||
async def async_await_connection(self) -> None:
|
||||
"""Wait for first connect."""
|
||||
await self._hub.event_connected.wait()
|
||||
await self.async_local_update(cancel_pending_update=True)
|
||||
|
||||
async def async_base_added_to_hass(self) -> None:
|
||||
"""Handle entity which will be added."""
|
||||
self.async_on_remove(
|
||||
async_call_later(
|
||||
self.hass,
|
||||
self._hub.config_delay + 0.1,
|
||||
self.async_await_connection,
|
||||
)
|
||||
# also runs when the add is aborted after the first update is scheduled,
|
||||
# and a rename removes and re-adds the entity, so it must not stop it
|
||||
self.async_on_remove(self._async_cancel_updates)
|
||||
self._cancel_call = self._async_call_later(
|
||||
self._hub.config_delay + 0.1, self.async_await_connection
|
||||
)
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
@@ -345,8 +376,8 @@ class ModbusToggleEntity(ModbusBaseEntity, ToggleEntity, RestoreEntity):
|
||||
if self._cancel_call:
|
||||
self._cancel_call()
|
||||
self._cancel_call = None
|
||||
self._cancel_call = async_call_later(
|
||||
self.hass, self._verify_delay, self.async_update
|
||||
self._cancel_call = self._async_call_later(
|
||||
self._verify_delay, self.async_update
|
||||
)
|
||||
return
|
||||
await self.async_local_update(cancel_pending_update=True)
|
||||
|
||||
@@ -168,7 +168,7 @@ async def mock_modbus_fixture(
|
||||
assert hub.event_connected.is_set()
|
||||
freezer.tick(timedelta(seconds=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
return mock_pymodbus
|
||||
|
||||
|
||||
@@ -218,7 +218,7 @@ async def mock_do_cycle_fixture(
|
||||
"""Trigger update call with time_changed event."""
|
||||
freezer.tick(timedelta(seconds=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
return freezer
|
||||
|
||||
|
||||
@@ -228,7 +228,7 @@ async def do_next_cycle(
|
||||
"""Trigger update call with time_changed event."""
|
||||
freezer.tick(timedelta(seconds=cycle))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
|
||||
@pytest.fixture(name="mock_test_state")
|
||||
|
||||
@@ -13,6 +13,8 @@ This file is responsible for testing:
|
||||
It uses binary_sensors/sensors to do black box testing of the read calls.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Awaitable, Callable
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from unittest import mock
|
||||
@@ -93,6 +95,7 @@ from homeassistant.components.modbus.validators import (
|
||||
)
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_STATE,
|
||||
CONF_ADDRESS,
|
||||
CONF_BINARY_SENSORS,
|
||||
@@ -109,6 +112,7 @@ from homeassistant.const import (
|
||||
CONF_STRUCTURE,
|
||||
CONF_TIMEOUT,
|
||||
CONF_TYPE,
|
||||
CONF_UNIQUE_ID,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
SERVICE_RELOAD,
|
||||
STATE_ON,
|
||||
@@ -116,7 +120,7 @@ from homeassistant.const import (
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
from homeassistant.helpers import entity_registry as er, issue_registry as ir
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
@@ -1061,10 +1065,10 @@ async def mock_modbus_read_pymodbus_fixture(
|
||||
assert caplog.text == ""
|
||||
freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL + 60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL + 60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
return mock_pymodbus
|
||||
|
||||
|
||||
@@ -1173,6 +1177,234 @@ async def test_pymodbus_close_fail(
|
||||
# Close() is called as part of teardown
|
||||
|
||||
|
||||
async def test_unreachable_device_does_not_hold_startup(
|
||||
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock
|
||||
) -> None:
|
||||
"""Test entities waiting for a device that never connects do not hold up startup."""
|
||||
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
|
||||
config = {
|
||||
DOMAIN: [
|
||||
{
|
||||
CONF_TYPE: TCP,
|
||||
CONF_HOST: TEST_MODBUS_HOST,
|
||||
CONF_PORT: TEST_PORT_TCP,
|
||||
CONF_NAME: TEST_MODBUS_NAME,
|
||||
CONF_SENSORS: [
|
||||
{
|
||||
CONF_NAME: TEST_ENTITY_NAME,
|
||||
CONF_ADDRESS: 51,
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_pymodbus.connect.return_value = False
|
||||
assert await async_setup_component(hass, DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# the entities start waiting for the first connection
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
|
||||
|
||||
# startup wraps up by waiting for foreground tasks, that wait must not be one
|
||||
async with asyncio.timeout(1):
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_aborted_add_cancels_first_update(
|
||||
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test an entity whose add is aborted does not run its first update later."""
|
||||
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
|
||||
config = {
|
||||
DOMAIN: [
|
||||
{
|
||||
CONF_TYPE: TCP,
|
||||
CONF_HOST: TEST_MODBUS_HOST,
|
||||
CONF_PORT: TEST_PORT_TCP,
|
||||
CONF_NAME: TEST_MODBUS_NAME,
|
||||
CONF_SENSORS: [
|
||||
{
|
||||
CONF_NAME: TEST_ENTITY_NAME,
|
||||
CONF_ADDRESS: 51,
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
# the platform aborts the add when restoring the last state fails, which
|
||||
# happens after the first update has been scheduled
|
||||
with mock.patch(
|
||||
"homeassistant.components.modbus.sensor.ModbusRegisterSensor.async_get_last_sensor_data",
|
||||
side_effect=ValueError("restore failed"),
|
||||
):
|
||||
assert await async_setup_component(hass, DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
caplog.clear()
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
assert not [record for record in caplog.records if record.levelno >= logging.ERROR]
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
|
||||
async def test_renamed_entity_keeps_polling(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
mock_pymodbus: mock.AsyncMock,
|
||||
) -> None:
|
||||
"""Test an entity polls again after a rename, which removes and re-adds it."""
|
||||
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
|
||||
config = {
|
||||
DOMAIN: [
|
||||
{
|
||||
CONF_TYPE: TCP,
|
||||
CONF_HOST: TEST_MODBUS_HOST,
|
||||
CONF_PORT: TEST_PORT_TCP,
|
||||
CONF_NAME: TEST_MODBUS_NAME,
|
||||
CONF_SENSORS: [
|
||||
{
|
||||
CONF_NAME: TEST_ENTITY_NAME,
|
||||
CONF_ADDRESS: 51,
|
||||
CONF_UNIQUE_ID: "renamed_sensor",
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
assert await async_setup_component(hass, DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert hass.states.get(entity_id).state == "0"
|
||||
|
||||
new_entity_id = f"{SENSOR_DOMAIN}.renamed"
|
||||
entity_registry.async_update_entity(entity_id, new_entity_id=new_entity_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x2A])
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=2))
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert hass.states.get(new_entity_id).state == "42"
|
||||
|
||||
|
||||
async def test_overlapping_updates_keep_one_poll_timer(
|
||||
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock
|
||||
) -> None:
|
||||
"""Test an update overlapping a poll does not start a second polling loop."""
|
||||
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
|
||||
config = {
|
||||
DOMAIN: [
|
||||
{
|
||||
CONF_TYPE: TCP,
|
||||
CONF_HOST: TEST_MODBUS_HOST,
|
||||
CONF_PORT: TEST_PORT_TCP,
|
||||
CONF_NAME: TEST_MODBUS_NAME,
|
||||
CONF_SENSORS: [
|
||||
{
|
||||
CONF_NAME: TEST_ENTITY_NAME,
|
||||
CONF_ADDRESS: 51,
|
||||
CONF_SCAN_INTERVAL: 10,
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
assert await async_setup_component(hass, "homeassistant", {})
|
||||
assert await async_setup_component(hass, DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
start = dt_util.utcnow()
|
||||
async_fire_time_changed(hass, start + timedelta(seconds=1))
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert hass.states.get(entity_id).state == "0"
|
||||
|
||||
# hold the next poll on the device, so an update requested now overlaps it
|
||||
release_poll = asyncio.Event()
|
||||
result = mock_pymodbus.read_holding_registers.return_value
|
||||
|
||||
async def _held_read(*args: object, **kwargs: object) -> ReadResult:
|
||||
await release_poll.wait()
|
||||
return result
|
||||
|
||||
mock_pymodbus.read_holding_registers.side_effect = _held_read
|
||||
async_fire_time_changed(hass, start + timedelta(seconds=12))
|
||||
await asyncio.sleep(0)
|
||||
await hass.services.async_call(
|
||||
"homeassistant", "update_entity", {ATTR_ENTITY_ID: entity_id}
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
release_poll.set()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
mock_pymodbus.read_holding_registers.reset_mock()
|
||||
async_fire_time_changed(hass, start + timedelta(seconds=25))
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert mock_pymodbus.read_holding_registers.call_count == 1
|
||||
|
||||
|
||||
async def _fire_first_connect_timer(hass: HomeAssistant) -> None:
|
||||
"""Let the entities start waiting for the first connection."""
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def _stop_hub(hass: HomeAssistant) -> None:
|
||||
"""Stop the hub, which also releases every wait for its connection."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN, SERVICE_STOP, {ATTR_HUB: TEST_MODBUS_NAME}, blocking=True
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"steps",
|
||||
[
|
||||
pytest.param(
|
||||
[_stop_hub, _fire_first_connect_timer], id="stopped_before_the_wait"
|
||||
),
|
||||
pytest.param(
|
||||
[_fire_first_connect_timer, _stop_hub], id="stopped_while_waiting"
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_stop_cancels_pending_first_update(
|
||||
hass: HomeAssistant,
|
||||
mock_pymodbus: mock.AsyncMock,
|
||||
steps: list[Callable[[HomeAssistant], Awaitable[None]]],
|
||||
) -> None:
|
||||
"""Test stopping the hub cancels an update still waiting for the first connection."""
|
||||
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
|
||||
config = {
|
||||
DOMAIN: [
|
||||
{
|
||||
CONF_TYPE: TCP,
|
||||
CONF_HOST: TEST_MODBUS_HOST,
|
||||
CONF_PORT: TEST_PORT_TCP,
|
||||
CONF_NAME: TEST_MODBUS_NAME,
|
||||
CONF_SENSORS: [
|
||||
{
|
||||
CONF_NAME: TEST_ENTITY_NAME,
|
||||
CONF_ADDRESS: 51,
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_pymodbus.connect.return_value = False
|
||||
assert await async_setup_component(hass, DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
|
||||
for step in steps:
|
||||
await step(hass)
|
||||
|
||||
# an update that survived the stop would run against the closed hub and
|
||||
# write the entity unavailable
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert hass.states.get(entity_id).state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_pymodbus_connect_fail(
|
||||
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, mock_pymodbus
|
||||
) -> None:
|
||||
@@ -1247,7 +1479,9 @@ async def test_delay(
|
||||
freezer.tick(timedelta(seconds=1, microseconds=999999))
|
||||
now = dt_util.utcnow()
|
||||
async_fire_time_changed(hass, now)
|
||||
await hass.async_block_till_done()
|
||||
# the hub sleeps the delay in a background task, so waiting for those
|
||||
# before the delay has passed would wait forever
|
||||
await hass.async_block_till_done(wait_background_tasks=now > time_after_delay)
|
||||
if now > time_sensor_active:
|
||||
if now <= time_after_delay:
|
||||
assert hass.states.get(entity_id).state in (
|
||||
@@ -1325,7 +1559,7 @@ async def test_integration_reload(
|
||||
caplog.clear()
|
||||
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
yaml_path = get_fixture_path("configuration.yaml", DOMAIN)
|
||||
with mock.patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
|
||||
|
||||
@@ -552,7 +552,7 @@ async def test_delay_switch(hass: HomeAssistant, mock_modbus) -> None:
|
||||
now = now + timedelta(seconds=2)
|
||||
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
||||
async_fire_time_changed(hass, now)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user