From 30522ed7d57e6f7fbbb23eeb117d270f7f343d59 Mon Sep 17 00:00:00 2001 From: David Mulcahey Date: Fri, 17 Jul 2026 12:27:52 -0400 Subject: [PATCH 1/2] Keep startup polling failures isolated --- tests/test_gateway.py | 202 +++++++++++++++++++++++++++++++++++++ zha/application/gateway.py | 72 +++++++++---- 2 files changed, 255 insertions(+), 19 deletions(-) diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 17792ba40..5727bff4b 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -59,6 +59,26 @@ } +def create_mains_powered_startup_device( + gateway: Gateway, + *, + ieee: str, + nwk: int, +) -> Device: + """Create a recent mains-powered device for startup polling tests.""" + zigpy_device = create_mock_zigpy_device( + gateway, + ZIGPY_DEVICE_BASIC, + ieee=ieee, + nwk=nwk, + ) + assert zigpy_device.node_desc is not None + zigpy_device.node_desc.mac_capability_flags |= ( + zigpy.zdo.types.NodeDescriptor.MACCapabilityFlags.MainsPowered + ) + return gateway.get_or_create_device(zigpy_device) + + async def coordinator_mock(zha_gateway: Gateway) -> Device: """Test ZHA light platform.""" @@ -247,6 +267,188 @@ async def test_mains_devices_startup_polling_config( await zha_gateway.async_block_till_done() +async def test_startup_polling_waits_for_remaining_devices_after_failure( + zha_data: ZHAData, + zigpy_app_controller: ControllerApplication, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test one device failure does not detach the remaining startup polls.""" + zha_gateway = Gateway(zha_data) + zha_gateway.application_controller = zigpy_app_controller + + failing_device = create_mains_powered_startup_device( + zha_gateway, + ieee="11:22:33:44:00:00:00:01", + nwk=0x1001, + ) + cancelled_device = create_mains_powered_startup_device( + zha_gateway, + ieee="11:22:33:44:00:00:00:02", + nwk=0x1002, + ) + blocked_device = create_mains_powered_startup_device( + zha_gateway, + ieee="11:22:33:44:00:00:00:03", + nwk=0x1003, + ) + + device_failure_raised = asyncio.Event() + blocked_device_started = asyncio.Event() + allow_blocked_device_to_finish = asyncio.Event() + blocked_device_finished = asyncio.Event() + + async def initialize_failing_device(*, from_cache: bool) -> None: + assert from_cache is False + device_failure_raised.set() + raise RuntimeError("device startup refresh failed") + + async def initialize_cancelled_device(*, from_cache: bool) -> None: + assert from_cache is False + raise asyncio.CancelledError + + async def initialize_blocked_device(*, from_cache: bool) -> None: + assert from_cache is False + blocked_device_started.set() + await allow_blocked_device_to_finish.wait() + blocked_device_finished.set() + + with ( + patch.object( + failing_device, + "async_initialize", + side_effect=initialize_failing_device, + ) as failing_device_initialize, + patch.object( + cancelled_device, + "async_initialize", + side_effect=initialize_cancelled_device, + ) as cancelled_device_initialize, + patch.object( + blocked_device, + "async_initialize", + side_effect=initialize_blocked_device, + ) as blocked_device_initialize, + patch.object( + type(zha_gateway), + "radio_concurrency", + new_callable=PropertyMock, + return_value=8, + ), + ): + startup_polling_task = asyncio.create_task( + zha_gateway.async_fetch_updated_state_mains() + ) + try: + await asyncio.gather( + device_failure_raised.wait(), + blocked_device_started.wait(), + ) + await asyncio.sleep(0) + + assert not startup_polling_task.done() + finally: + allow_blocked_device_to_finish.set() + await startup_polling_task + + assert blocked_device_finished.is_set() + failing_device_initialize.assert_awaited_once_with(from_cache=False) + cancelled_device_initialize.assert_awaited_once_with(from_cache=False) + blocked_device_initialize.assert_awaited_once_with(from_cache=False) + assert ( + f"[{failing_device.nwk}]({failing_device.name}) " + "Failed to refresh state during startup polling" + ) in caplog.text + assert "RuntimeError: device startup refresh failed" in caplog.text + assert ( + f"[{cancelled_device.nwk}]({cancelled_device.name}) " + "Startup state refresh was cancelled" + ) in caplog.text + + +async def test_startup_polling_restores_polling_after_unexpected_failure( + zha_data: ZHAData, + zigpy_app_controller: ControllerApplication, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test polling is restored after an unexpected startup polling failure.""" + zha_gateway = Gateway(zha_data) + zha_gateway.application_controller = zigpy_app_controller + + with patch.object( + zha_gateway, + "async_fetch_updated_state_mains", + side_effect=RuntimeError("unexpected startup polling failure"), + ) as fetch_updated_state_mains: + await zha_gateway.async_initialize_devices_and_entities() + startup_polling_task = next( + task + for task in zha_gateway._background_tasks + if task.get_name() == "zha.gateway-fetch_updated_state" + ) + await startup_polling_task + + fetch_updated_state_mains.assert_awaited_once_with() + assert zha_gateway.config.allow_polling is True + assert "Unexpected failure during startup polling" in caplog.text + assert "RuntimeError: unexpected startup polling failure" in caplog.text + + +async def test_startup_polling_cancellation_propagates( + zha_data: ZHAData, + zigpy_app_controller: ControllerApplication, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test cancelling startup polling cancels its device initialization.""" + zha_gateway = Gateway(zha_data) + zha_gateway.application_controller = zigpy_app_controller + + blocked_device = create_mains_powered_startup_device( + zha_gateway, + ieee="11:22:33:44:00:00:00:04", + nwk=0x1004, + ) + blocked_device_started = asyncio.Event() + blocked_device_cancelled = asyncio.Event() + never_finish = asyncio.Event() + + async def initialize_blocked_device(*, from_cache: bool) -> None: + if from_cache: + return + blocked_device_started.set() + try: + await never_finish.wait() + except asyncio.CancelledError: + blocked_device_cancelled.set() + raise + + with patch.object( + blocked_device, + "async_initialize", + side_effect=initialize_blocked_device, + ): + await zha_gateway.async_initialize_devices_and_entities() + startup_polling_task = next( + task + for task in zha_gateway._background_tasks + if task.get_name() == "zha.gateway-fetch_updated_state" + ) + try: + await blocked_device_started.wait() + + startup_polling_task.cancel() + with pytest.raises(asyncio.CancelledError): + await startup_polling_task + finally: + if not startup_polling_task.done(): + startup_polling_task.cancel() + with suppress(asyncio.CancelledError): + await startup_polling_task + + assert blocked_device_cancelled.is_set() + assert zha_gateway.config.allow_polling is True + assert "Unexpected failure during startup polling" not in caplog.text + + async def test_gateway_group_methods( zha_gateway: Gateway, caplog: pytest.LogCaptureFixture, diff --git a/zha/application/gateway.py b/zha/application/gateway.py index ecff916d1..8411a1354 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -352,25 +352,51 @@ async def async_fetch_updated_state_mains(self) -> None: # Only delay startup to poll mains-powered devices that are online online_devices = [ - dev - for dev in self.devices.values() - if dev.is_mains_powered - and dev.last_seen is not None - and (now - dev.last_seen) < dev.consider_unavailable_time + device + for device in self.devices.values() + if device.is_mains_powered + and device.last_seen is not None + and (now - device.last_seen) < device.consider_unavailable_time ] # Prioritize devices that have recently been contacted - online_devices.sort(key=lambda dev: cast(float, dev.last_seen), reverse=True) + online_devices.sort( + key=lambda device: cast(float, device.last_seen), reverse=True + ) # Make sure that we always leave slots for non-startup requests - max_poll_concurrency = max(1, self.radio_concurrency - 4) - - await gather_with_limited_concurrency( - max_poll_concurrency, - *(dev.async_initialize(from_cache=False) for dev in online_devices), + maximum_polling_concurrency = max(1, self.radio_concurrency - 4) + + # Startup polling is best effort. Wait for every device so a failure cannot + # leave sibling initialization tasks running after this operation completes. + initialization_results = await gather_with_limited_concurrency( + maximum_polling_concurrency, + *(device.async_initialize(from_cache=False) for device in online_devices), + return_exceptions=True, ) - _LOGGER.debug("completed fetching current state for mains powered devices") + for device, initialization_result in zip( + online_devices, initialization_results, strict=True + ): + if isinstance(initialization_result, asyncio.CancelledError): + _LOGGER.debug( + "[%s](%s) Startup state refresh was cancelled", + device.nwk, + device.name, + ) + elif isinstance(initialization_result, Exception): + _LOGGER.warning( + "[%s](%s) Failed to refresh state during startup polling", + device.nwk, + device.name, + exc_info=( + type(initialization_result), + initialization_result, + initialization_result.__traceback__, + ), + ) + + _LOGGER.debug("Completed fetching current state for mains powered devices") async def async_initialize_devices_and_entities(self) -> None: """Initialize devices and load entities.""" @@ -382,13 +408,21 @@ async def async_initialize_devices_and_entities(self) -> None: async def fetch_updated_state() -> None: """Fetch updated state for mains powered devices.""" - if self.config.config.device_options.enable_mains_startup_polling: - async with self.request_priority(t.PacketPriority.LOW): - await self.async_fetch_updated_state_mains() - else: - _LOGGER.debug("Polling of mains powered devices at startup is disabled") - _LOGGER.debug("Allowing polled requests") - self.config.allow_polling = True + try: + if self.config.config.device_options.enable_mains_startup_polling: + async with self.request_priority(t.PacketPriority.LOW): + await self.async_fetch_updated_state_mains() + else: + _LOGGER.debug( + "Polling of mains powered devices at startup is disabled" + ) + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.exception("Unexpected failure during startup polling") + finally: + self.config.allow_polling = True + _LOGGER.debug("Allowing polled requests") # background the fetching of state for mains powered devices self.async_create_background_task( From 28b6884941d917c4d369ccfe00d3d2c2d248d6f1 Mon Sep 17 00:00:00 2001 From: David Mulcahey Date: Fri, 17 Jul 2026 14:42:59 -0400 Subject: [PATCH 2/2] Delegate startup polling concurrency to zigpy --- tests/test_device.py | 55 ++++++++++++ tests/test_gateway.py | 168 +++++++++++++++++++---------------- zha/application/gateway.py | 40 ++++----- zha/zigbee/cluster_config.py | 18 +++- zha/zigbee/device.py | 15 +++- 5 files changed, 191 insertions(+), 105 deletions(-) diff --git a/tests/test_device.py b/tests/test_device.py index 8c9703555..9824bb19e 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -57,6 +57,7 @@ from zha.application.platforms.switch import Switch from zha.exceptions import ZHAException from zha.quirks import DeviceRegistry +from zha.zigbee.cluster_config import AggregatedAttrConfig, AggregatedClusterConfig from zha.zigbee.device import ( ClusterBinding, Device, @@ -124,6 +125,60 @@ def zigpy_device_mains(zha_gateway: Gateway, with_basic_cluster: bool = True): ) +@pytest.mark.parametrize( + ("request_priority", "expected_read_attribute_options"), + [ + (None, {}), + ( + zigpy.types.PacketPriority.LOW, + {"priority": zigpy.types.PacketPriority.LOW}, + ), + ], +) +async def test_initialize_request_priority( + zha_gateway: Gateway, + request_priority: int | None, + expected_read_attribute_options: dict[str, int], +) -> None: + """Test device initialization forwards request priority to attribute reads.""" + zha_device = zha_gateway.get_or_create_device(zigpy_device_mains(zha_gateway)) + cluster = mock.MagicMock() + cluster.read_attributes = AsyncMock() + cluster_configs = { + (1, 6, True): AggregatedClusterConfig( + cluster=cluster, + attributes={ + "cached_attribute": AggregatedAttrConfig(read_on_startup=False), + "fresh_attribute": AggregatedAttrConfig(read_on_startup=True), + }, + ) + } + + with patch( + "zha.zigbee.device.aggregate_cluster_configs", + return_value=cluster_configs, + ): + await zha_device.async_initialize( + from_cache=False, + request_priority=request_priority, + ) + + assert cluster.read_attributes.await_args_list == [ + call( + ["cached_attribute"], + allow_cache=True, + only_cache=False, + **expected_read_attribute_options, + ), + call( + ["fresh_attribute"], + allow_cache=False, + only_cache=False, + **expected_read_attribute_options, + ), + ] + + async def _send_time_changed(zha_gateway: Gateway, seconds: int): """Send a time changed event.""" await asyncio.sleep(seconds) diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 5727bff4b..6856e8a6e 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -2,7 +2,7 @@ import asyncio from contextlib import suppress -from unittest.mock import AsyncMock, MagicMock, PropertyMock, call, patch +from unittest.mock import AsyncMock, MagicMock, call, patch import pytest from zhaquirks.builder import QuirkBuilder @@ -297,17 +297,33 @@ async def test_startup_polling_waits_for_remaining_devices_after_failure( allow_blocked_device_to_finish = asyncio.Event() blocked_device_finished = asyncio.Event() - async def initialize_failing_device(*, from_cache: bool) -> None: + def assert_low_startup_priority(request_priority: int | None) -> None: + assert request_priority == zigpy.types.PacketPriority.LOW + assert ( + zha_gateway.application_controller._packet_priority_var.get() + == zigpy.types.PacketPriority.LOW + ) # pylint: disable=protected-access + + async def initialize_failing_device( + *, from_cache: bool, request_priority: int | None + ) -> None: assert from_cache is False + assert_low_startup_priority(request_priority) device_failure_raised.set() raise RuntimeError("device startup refresh failed") - async def initialize_cancelled_device(*, from_cache: bool) -> None: + async def initialize_cancelled_device( + *, from_cache: bool, request_priority: int | None + ) -> None: assert from_cache is False + assert_low_startup_priority(request_priority) raise asyncio.CancelledError - async def initialize_blocked_device(*, from_cache: bool) -> None: + async def initialize_blocked_device( + *, from_cache: bool, request_priority: int | None + ) -> None: assert from_cache is False + assert_low_startup_priority(request_priority) blocked_device_started.set() await allow_blocked_device_to_finish.wait() blocked_device_finished.set() @@ -328,12 +344,6 @@ async def initialize_blocked_device(*, from_cache: bool) -> None: "async_initialize", side_effect=initialize_blocked_device, ) as blocked_device_initialize, - patch.object( - type(zha_gateway), - "radio_concurrency", - new_callable=PropertyMock, - return_value=8, - ), ): startup_polling_task = asyncio.create_task( zha_gateway.async_fetch_updated_state_mains() @@ -351,9 +361,18 @@ async def initialize_blocked_device(*, from_cache: bool) -> None: await startup_polling_task assert blocked_device_finished.is_set() - failing_device_initialize.assert_awaited_once_with(from_cache=False) - cancelled_device_initialize.assert_awaited_once_with(from_cache=False) - blocked_device_initialize.assert_awaited_once_with(from_cache=False) + failing_device_initialize.assert_awaited_once_with( + from_cache=False, + request_priority=zigpy.types.PacketPriority.LOW, + ) + cancelled_device_initialize.assert_awaited_once_with( + from_cache=False, + request_priority=zigpy.types.PacketPriority.LOW, + ) + blocked_device_initialize.assert_awaited_once_with( + from_cache=False, + request_priority=zigpy.types.PacketPriority.LOW, + ) assert ( f"[{failing_device.nwk}]({failing_device.name}) " "Failed to refresh state during startup polling" @@ -411,9 +430,17 @@ async def test_startup_polling_cancellation_propagates( blocked_device_cancelled = asyncio.Event() never_finish = asyncio.Event() - async def initialize_blocked_device(*, from_cache: bool) -> None: + async def initialize_blocked_device( + *, from_cache: bool, request_priority: int | None = None + ) -> None: if from_cache: + assert request_priority is None return + assert request_priority == zigpy.types.PacketPriority.LOW + assert ( + zha_gateway.application_controller._packet_priority_var.get() + == zigpy.types.PacketPriority.LOW + ) # pylint: disable=protected-access blocked_device_started.set() try: await never_finish.wait() @@ -638,81 +665,68 @@ async def test_remove_device_cleans_up_group_membership( assert device_light_1.ieee not in zha_gateway.devices -@pytest.mark.parametrize("radio_concurrency", [1, 2, 8]) -async def test_startup_concurrency_limit( - radio_concurrency: int, +async def test_startup_polling_uses_low_priority_without_limiting_initialization( zigpy_app_controller: ControllerApplication, zha_data: ZHAData, -): - """Test ZHA gateway limits concurrency on startup.""" - zha_gw = Gateway(zha_data) +) -> None: + """Test all startup initializers run with low request priority.""" + zha_gateway = Gateway(zha_data) + zha_gateway.application_controller = zigpy_app_controller - with patch( - "bellows.zigbee.application.ControllerApplication.new", - return_value=zigpy_app_controller, - ): - await zha_gw.async_initialize() - - for i in range(50): - zigpy_dev = create_mock_zigpy_device( - zha_gw, - { - 1: { - SIG_EP_INPUT: [ - general.OnOff.cluster_id, - general.LevelControl.cluster_id, - lighting.Color.cluster_id, - general.Groups.cluster_id, - ], - SIG_EP_OUTPUT: [], - SIG_EP_TYPE: zha.DeviceType.COLOR_DIMMABLE_LIGHT, - SIG_EP_PROFILE: zha.PROFILE_ID, - } - }, - ieee=f"11:22:33:44:{i:08x}", - nwk=0x1234 + i, + device_count = 8 + for device_number in range(device_count): + create_mains_powered_startup_device( + zha_gateway, + ieee=f"11:22:33:44:{device_number:08x}", + nwk=0x1234 + device_number, ) - zigpy_dev.node_desc.mac_capability_flags |= ( - zigpy.zdo.types.NodeDescriptor.MACCapabilityFlags.MainsPowered - ) - - zha_gw.get_or_create_device(zigpy_dev) - - # Keep track of request concurrency during initialization - current_concurrency = 0 - concurrencies = [] - async def mock_send_packet(*args, **kwargs): # pylint: disable=unused-argument - """Mock send packet.""" - nonlocal current_concurrency + active_initialization_count = 0 + maximum_active_initialization_count = 0 + all_initializations_started = asyncio.Event() + allow_initializations_to_finish = asyncio.Event() - current_concurrency += 1 - concurrencies.append(current_concurrency) + async def initialize_device( + *, from_cache: bool, request_priority: int | None + ) -> None: + nonlocal active_initialization_count, maximum_active_initialization_count - await asyncio.sleep(0.001) - - current_concurrency -= 1 - concurrencies.append(current_concurrency) + assert from_cache is False + assert request_priority == zigpy.types.PacketPriority.LOW + assert ( + zha_gateway.application_controller._packet_priority_var.get() + == zigpy.types.PacketPriority.LOW + ) # pylint: disable=protected-access + + active_initialization_count += 1 + maximum_active_initialization_count = max( + maximum_active_initialization_count, + active_initialization_count, + ) + if active_initialization_count == device_count: + all_initializations_started.set() - type(zha_gw).radio_concurrency = PropertyMock(return_value=radio_concurrency) - assert zha_gw.radio_concurrency == radio_concurrency + try: + await allow_initializations_to_finish.wait() + finally: + active_initialization_count -= 1 with patch( "zha.zigbee.device.Device.async_initialize", - side_effect=mock_send_packet, - ): - await zha_gw.async_fetch_updated_state_mains() - - await zha_gw.shutdown() - - # Make sure concurrency was always limited - assert current_concurrency == 0 - assert min(concurrencies) == 0 + side_effect=initialize_device, + ) as initialize_device_mock: + startup_polling_task = asyncio.create_task( + zha_gateway.async_fetch_updated_state_mains() + ) + try: + await all_initializations_started.wait() + finally: + allow_initializations_to_finish.set() + await startup_polling_task - if radio_concurrency > 1: - assert 1 <= max(concurrencies) < zha_gw.radio_concurrency - else: - assert 1 == max(concurrencies) == zha_gw.radio_concurrency + assert initialize_device_mock.await_count == device_count + assert maximum_active_initialization_count == device_count + assert active_initialization_count == 0 async def test_gateway_device_removed(zha_gateway: Gateway) -> None: diff --git a/zha/application/gateway.py b/zha/application/gateway.py index 8411a1354..816aa8022 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -47,11 +47,7 @@ RadioType, ) from zha.application.helpers import DeviceAvailabilityChecker, GlobalUpdater, ZHAData -from zha.async_ import ( - AsyncUtilMixin, - create_eager_task, - gather_with_limited_concurrency, -) +from zha.async_ import AsyncUtilMixin, create_eager_task from zha.event import EventBase from zha.quirks import DEVICE_REGISTRY, QUIRK_REGISTRY_ENTRY_ATTR from zha.zigbee.device import Device, DeviceInfo, DeviceStatus, ExtendedDeviceInfo @@ -337,13 +333,6 @@ def load_groups(self) -> None: for entity in discovery.discover_group_entities(zha_group): entity.on_add() - @property - def radio_concurrency(self) -> int: - """Maximum configured radio concurrency.""" - return ( - self.application_controller._concurrent_requests_semaphore.max_concurrency - ) # pylint: disable=protected-access - async def async_fetch_updated_state_mains(self) -> None: """Fetch updated state for mains powered devices.""" _LOGGER.debug("Fetching current state for mains powered devices") @@ -364,16 +353,24 @@ async def async_fetch_updated_state_mains(self) -> None: key=lambda device: cast(float, device.last_seen), reverse=True ) - # Make sure that we always leave slots for non-startup requests - maximum_polling_concurrency = max(1, self.radio_concurrency - 4) - # Startup polling is best effort. Wait for every device so a failure cannot # leave sibling initialization tasks running after this operation completes. - initialization_results = await gather_with_limited_concurrency( - maximum_polling_concurrency, - *(device.async_initialize(from_cache=False) for device in online_devices), - return_exceptions=True, - ) + startup_polling_priority = t.PacketPriority.LOW + + # Keep the ambient priority for radio libraries that resolve the request + # context, and pass it explicitly for libraries such as zigpy-ziggurat that + # transmit the priority stored on each packet. + async with self.request_priority(startup_polling_priority): + initialization_results = await asyncio.gather( + *( + device.async_initialize( + from_cache=False, + request_priority=startup_polling_priority, + ) + for device in online_devices + ), + return_exceptions=True, + ) for device, initialization_result in zip( online_devices, initialization_results, strict=True @@ -410,8 +407,7 @@ async def fetch_updated_state() -> None: """Fetch updated state for mains powered devices.""" try: if self.config.config.device_options.enable_mains_startup_polling: - async with self.request_priority(t.PacketPriority.LOW): - await self.async_fetch_updated_state_mains() + await self.async_fetch_updated_state_mains() else: _LOGGER.debug( "Polling of mains powered devices at startup is disabled" diff --git a/zha/zigbee/cluster_config.py b/zha/zigbee/cluster_config.py index f0d487d01..38f947725 100644 --- a/zha/zigbee/cluster_config.py +++ b/zha/zigbee/cluster_config.py @@ -245,8 +245,14 @@ async def configure_cluster_configs( async def initialize_cluster_configs( configs: dict[tuple[int, int, bool], AggregatedClusterConfig], from_cache: bool, + *, + request_priority: int | None = None, ) -> None: - """Read initial attribute values from aggregated configs.""" + """Read initial attribute values with an optional request priority.""" + read_attribute_options = ( + {"priority": request_priority} if request_priority is not None else {} + ) + for agg in configs.values(): cached_attrs = [ attr_name @@ -262,7 +268,10 @@ async def initialize_cluster_configs( if cached_attrs: try: await agg.cluster.read_attributes( - cached_attrs, allow_cache=True, only_cache=from_cache + cached_attrs, + allow_cache=True, + only_cache=from_cache, + **read_attribute_options, ) except Exception as ex: # pylint: disable=broad-except _LOGGER.debug( @@ -276,7 +285,10 @@ async def initialize_cluster_configs( if fresh_attrs: try: await agg.cluster.read_attributes( - fresh_attrs, allow_cache=from_cache, only_cache=from_cache + fresh_attrs, + allow_cache=from_cache, + only_cache=from_cache, + **read_attribute_options, ) except Exception as ex: # pylint: disable=broad-except _LOGGER.debug( diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index d9a6a1cb0..7700c52d3 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -1204,8 +1204,13 @@ async def recompute_entities(self) -> None: self._discover_new_entities() await self._add_pending_entities() - async def async_initialize(self, from_cache: bool = False) -> None: - """Initialize cluster handlers.""" + async def async_initialize( + self, + from_cache: bool = False, + *, + request_priority: int | None = None, + ) -> None: + """Initialize cluster handlers with an optional request priority.""" self.debug("started initialization") # We discover prospective entities before initialization @@ -1214,7 +1219,11 @@ async def async_initialize(self, from_cache: bool = False) -> None: # Read initial attributes from entity-level cluster configs aggregated = aggregate_cluster_configs(self._discovered_entities) if aggregated and not self.skip_configuration: - await initialize_cluster_configs(aggregated, from_cache) + await initialize_cluster_configs( + aggregated, + from_cache, + request_priority=request_priority, + ) # And add them after. Emit events only on re-initialization, not the first. await self._add_pending_entities(emit_event=self._initialized)