diff --git a/tests/test_light.py b/tests/test_light.py index 448e3f68b..177adeaaa 100644 --- a/tests/test_light.py +++ b/tests/test_light.py @@ -2069,6 +2069,113 @@ async def test_turn_on_during_off_transition(zha_gateway: Gateway) -> None: assert bool(entity.state.on) is False +@patch( + "zigpy.zcl.clusters.general.LevelControl.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +@patch( + "zigpy.zcl.clusters.general.OnOff.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +async def test_zero_level_report_during_transition_turns_light_off( + zha_gateway: Gateway, +) -> None: + """Test a current_level=0 report buffered during a transition ends the light off. + + A device that reports level 0 within the transition window is actually off, + so the optimistic on-state is overridden when the window ends. + """ + device_light_1 = await device_light_1_mock(zha_gateway) + dev1_cluster_level = device_light_1.device.endpoints[1].level + entity = get_entity(device_light_1, platform=Platform.LIGHT) + + await entity.async_turn_on(brightness=200) + await zha_gateway.async_block_till_done() + assert bool(entity.state.on) is True + assert entity.is_transitioning + + await send_attributes_report( + zha_gateway, + dev1_cluster_level, + {general.LevelControl.AttributeDefs.current_level.id: 0}, + ) + await zha_gateway.async_block_till_done() + + # Still optimistically on while the window is open. + assert bool(entity.state.on) is True + + await asyncio.sleep(1.6) + await zha_gateway.async_block_till_done() + assert not entity.is_transitioning + assert bool(entity.state.on) is False + + +@patch( + "zigpy.zcl.clusters.general.LevelControl.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +@patch( + "zigpy.zcl.clusters.general.OnOff.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +@pytest.mark.parametrize( + ("on_off_reports", "expected_on"), + [ + ([1], True), + ([1, 0], False), + ], +) +async def test_on_off_report_during_off_transition_is_applied( + zha_gateway: Gateway, on_off_reports: list[int], expected_on: bool +) -> None: + """Test an on_off report received during an off transition is applied when it ends. + + Something other than this entity (a bound remote, or a cluster command sent + through the ZHA service) can turn the light on within the transition window + of a turn_off. The device's on_off=True report must be buffered like an off + report, not discarded, or the entity stays off while the light is on. The + last report in the window wins. + """ + device_light_1 = await device_light_1_mock(zha_gateway) + dev1_cluster_on_off = device_light_1.device.endpoints[1].on_off + dev1_cluster_level = device_light_1.device.endpoints[1].level + entity = get_entity(device_light_1, platform=Platform.LIGHT) + + await entity.async_turn_on(brightness=200) + await zha_gateway.async_block_till_done() + await asyncio.sleep(1.6) # let the turn_on transition window end + await zha_gateway.async_block_till_done() + assert not entity.is_transitioning + + # Turn it off; the transition window opens (default 1 + 0.5s = 1.5s). + await entity.async_turn_off() + await zha_gateway.async_block_till_done() + assert bool(entity.state.on) is False + assert entity.is_transitioning + + # Within the window, the light is turned on from outside this entity. + for value in on_off_reports: + await send_attributes_report( + zha_gateway, + dev1_cluster_on_off, + {general.OnOff.AttributeDefs.on_off.id: value}, + ) + await send_attributes_report( + zha_gateway, + dev1_cluster_level, + {general.LevelControl.AttributeDefs.current_level.id: 128}, + ) + await zha_gateway.async_block_till_done() + assert bool(entity.state.on) is False + + # Once the transition timer fires, the buffered reports are applied. + await asyncio.sleep(1.6) + await zha_gateway.async_block_till_done() + assert not entity.is_transitioning + assert bool(entity.state.on) is expected_on + assert entity.state.brightness == 128 + + @patch( "zigpy.zcl.clusters.general.OnOff.request", new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index 622cbae66..259d8a1e9 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -274,6 +274,7 @@ def __init__(self, *args, **kwargs): self._transitioning_group: bool = False self._transition_listener: asyncio.TimerHandle | None = None self._transition_brightness_buffer: int | None = None + self._transition_on_off_buffer: bool | None = None self._internal_supported_color_modes: set[ColorMode] = set() @@ -798,6 +799,7 @@ def async_transition_set_flag(self) -> None: self._transitioning_individual = True self._transitioning_group = False self._transition_brightness_buffer = None + self._transition_on_off_buffer = None if isinstance(self, LightGroup): for platform_entity in self.group.get_platform_entities(Light.PLATFORM): assert isinstance(platform_entity, Light) @@ -855,6 +857,16 @@ def async_transition_complete(self, _=None) -> None: else: self._brightness = self._transition_brightness_buffer self._transition_brightness_buffer = None + if self._transition_on_off_buffer is not None: + self.debug( + "applying buffered onoff %s from transition", + self._transition_on_off_buffer, + ) + self._state = self._transition_on_off_buffer + if self._state: + self._off_with_transition = False + self._off_brightness = None + self._transition_on_off_buffer = None self.maybe_emit_state_changed_event() if isinstance(self, LightGroup): for platform_entity in self.group.get_platform_entities(Light.PLATFORM): @@ -1186,8 +1198,7 @@ def handle_attribute_updated( "received onoff %s while transitioning - buffering", event.value, ) - if not event.value: - self._transition_brightness_buffer = 0 + self._transition_on_off_buffer = bool(event.value) return self._state = bool(event.value) if event.value: