From 79d6bf3d11599c6a0dce07cd94d4126562449ff9 Mon Sep 17 00:00:00 2001 From: David Mulcahey Date: Tue, 14 Jul 2026 09:41:08 -0400 Subject: [PATCH 1/2] Deactivate light color loops before color commands --- tests/test_light.py | 213 ++++++++++++++++++++ zha/application/platforms/light/__init__.py | 47 ++++- 2 files changed, 253 insertions(+), 7 deletions(-) diff --git a/tests/test_light.py b/tests/test_light.py index a7dcc4419..ba6497551 100644 --- a/tests/test_light.py +++ b/tests/test_light.py @@ -1838,6 +1838,219 @@ async def test_on_with_off_color(zha_gateway: Gateway) -> None: assert entity.state["color_mode"] == ColorMode.COLOR_TEMP +def _restore_colorloop_test_state( + entity: Any, + *, + state: bool, + effect: str = "colorloop", + off_with_transition: bool = False, + off_brightness: int | None = None, +) -> None: + """Restore a light to a known colorloop test state.""" + entity.restore_external_state_attributes( + state=state, + off_with_transition=off_with_transition, + off_brightness=off_brightness, + brightness=100, + color_temp=None, + xy_color=(0.5, 0.5), + color_mode=ColorMode.XY, + effect=effect, + ) + + +def _color_command_ids(cluster: Any) -> list[int]: + """Return sent color-cluster command IDs.""" + return [request.args[1] for request in cluster.request.await_args_list] + + +@pytest.mark.parametrize( + ("initial_state", "initial_effect", "color_kwargs", "expected_commands"), + [ + ( + True, + "colorloop", + {"color_temp": 235}, + ("color_loop_set", "move_to_color_temp"), + ), + ( + True, + "colorloop", + {"xy_color": (0.25, 0.5)}, + ("color_loop_set", "move_to_color"), + ), + (True, "off", {"color_temp": 235}, ("move_to_color_temp",)), + ( + False, + "colorloop", + {"color_temp": 235}, + ("move_to_color_temp", "color_loop_set"), + ), + ], +) +@patch( + "zigpy.zcl.clusters.lighting.Color.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +@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_turn_on_colorloop_deactivation_order_and_guard( + zha_gateway: Gateway, + initial_state: bool, + initial_effect: str, + color_kwargs: dict[str, Any], + expected_commands: tuple[str, ...], +) -> None: + """Test colorloop deactivation ordering and the inactive guard.""" + device_light = await device_light_3_mock(zha_gateway) + entity = get_entity(device_light, platform=Platform.LIGHT) + cluster_color = device_light.device.endpoints[1].light_color + + _restore_colorloop_test_state( + entity, + state=initial_state, + effect=initial_effect, + off_with_transition=not initial_state, + off_brightness=80 if not initial_state else None, + ) + cluster_color.request.reset_mock() + + await entity.async_turn_on(**color_kwargs) + + if "color_loop_set" in expected_commands: + command_index = expected_commands.index("color_loop_set") + assert ( + cluster_color.request.await_args_list[command_index].kwargs["action"] + is lighting.Color.ColorLoopAction.Deactivate + ) + assert _color_command_ids(cluster_color) == [ + cluster_color.commands_by_name[command].id for command in expected_commands + ] + assert entity.state["effect"] == "off" + assert entity.state["off_with_transition"] is False + assert entity.state["off_brightness"] is None + entity.async_transition_complete() + assert not entity.is_transitioning + + +@pytest.mark.parametrize( + "status", + [zcl_f.Status.FAILURE, zcl_f.Status.UNSUP_CLUSTER_COMMAND], +) +@patch( + "zigpy.zcl.clusters.lighting.Color.request", + new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]), +) +@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_colorloop_deactivation_failure_preserves_independent_commands( + zha_gateway: Gateway, + status: zcl_f.Status, +) -> None: + """Test deactivation failure blocks only the dependent color command.""" + device_light = await device_light_3_mock(zha_gateway) + entity = get_entity(device_light, platform=Platform.LIGHT) + cluster_color = device_light.device.endpoints[1].light_color + cluster_level = device_light.device.endpoints[1].level + + _restore_colorloop_test_state( + entity, + state=True, + off_with_transition=True, + off_brightness=80, + ) + cluster_color.request.reset_mock() + cluster_level.request.reset_mock() + cluster_color.request.return_value = [sentinel.data, status] + + await entity.async_turn_on(brightness=150, color_temp=235, effect="off") + + assert _color_command_ids(cluster_color) == [ + cluster_color.commands_by_name["color_loop_set"].id + ] + assert cluster_level.request.await_count == 1 + assert entity.state["on"] is True + assert entity.state["brightness"] == 150 + assert entity.state["effect"] == "colorloop" + assert entity.state["color_mode"] == ColorMode.XY + assert entity.state["color_temp"] is None + assert entity.state["xy_color"] == (0.5, 0.5) + assert entity.state["off_with_transition"] is False + assert entity.state["off_brightness"] is None + assert entity.is_transitioning + entity.async_transition_complete() + assert not entity.is_transitioning + + +@pytest.mark.parametrize( + ("device_factory", "status", "expected_effect", "sends_command"), + [ + (device_light_3_mock, zcl_f.Status.SUCCESS, "colorloop", True), + (device_light_3_mock, zcl_f.Status.FAILURE, "off", True), + ( + device_light_3_mock, + zcl_f.Status.UNSUP_CLUSTER_COMMAND, + "off", + True, + ), + (device_light_1_mock, zcl_f.Status.SUCCESS, "off", False), + ], +) +@patch( + "zigpy.zcl.clusters.lighting.Color.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_requested_colorloop_activation_policy( + zha_gateway: Gateway, + device_factory: Any, + status: zcl_f.Status, + expected_effect: str, + sends_command: bool, +) -> None: + """Test activation requires support and a successful command response.""" + device_light = await device_factory(zha_gateway) + entity = get_entity(device_light, platform=Platform.LIGHT) + cluster_color = device_light.device.endpoints[1].light_color + + _restore_colorloop_test_state( + entity, + state=True, + effect="off" if sends_command else "colorloop", + ) + assert entity.state["effect"] == "off" + cluster_color.request.reset_mock() + cluster_color.request.return_value = [sentinel.data, status] + + await entity.async_turn_on(effect="colorloop") + + expected_commands = ( + [cluster_color.commands_by_name["color_loop_set"].id] if sends_command else [] + ) + assert _color_command_ids(cluster_color) == expected_commands + if sends_command: + assert ( + cluster_color.request.await_args.kwargs["action"] + is lighting.Color.ColorLoopAction.Activate_from_current_hue + ) + assert entity.state["effect"] == expected_effect + + @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 c6f1249f4..93c2ad8aa 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -256,7 +256,12 @@ def restore_external_state_attributes( if color_mode is not None and color_mode in self._supported_color_modes: self._color_mode = color_mode if effect is not None: - self._effect = effect + self._effect = ( + EFFECT_OFF + if effect == EFFECT_COLORLOOP + and EFFECT_COLORLOOP not in (self._effect_list or ()) + else effect + ) class BaseSharedLight(BaseLight): @@ -518,6 +523,32 @@ async def _async_turn_on_impl( # noqa: C901 t_log = {} + color_loop_supported = EFFECT_COLORLOOP in (self._effect_list or ()) + if not color_loop_supported and self._effect == EFFECT_COLORLOOP: + self._effect = EFFECT_OFF + color_loop_remembered = ( + color_loop_supported and self._effect == EFFECT_COLORLOOP + ) + color_loop_active = self._state and color_loop_remembered + deactivate_color_loop = color_loop_supported and effect != EFFECT_COLORLOOP + dependent_color_commands_allowed = True + if deactivate_color_loop and color_loop_active: + assert self._color_cluster is not None + result = await self._color_cluster.color_loop_set( + update_flags=Color.ColorLoopUpdateFlags.Action, + action=Color.ColorLoopAction.Deactivate, + direction=Color.ColorLoopDirection.Decrement, + time=0, + start_hue=0, + ) + t_log["color_loop_set"] = result + dependent_color_commands_allowed = result[1] is Status.SUCCESS + if dependent_color_commands_allowed: + self._effect = EFFECT_OFF + deactivate_color_loop_after_turn_on = ( + deactivate_color_loop and color_loop_remembered and not color_loop_active + ) + if new_color_provided_while_off: assert self._level_cluster is not None @@ -541,7 +572,7 @@ async def _async_turn_on_impl( # noqa: C901 # be set at the second move_to_level call self._state = True - if execute_if_off_supported: + if execute_if_off_supported and dependent_color_commands_allowed: self.debug("handling color commands before turning on/level") if not await self.async_handle_color_commands( color_temp, @@ -602,7 +633,7 @@ async def _async_turn_on_impl( # noqa: C901 return self._state = True - if not execute_if_off_supported: + if not execute_if_off_supported and dependent_color_commands_allowed: self.debug("handling color commands after turning on/level") if not await self.async_handle_color_commands( color_temp, @@ -644,7 +675,7 @@ async def _async_turn_on_impl( # noqa: C901 self.async_transition_start_timer(transition_time) if self._color_cluster is not None: - if effect == EFFECT_COLORLOOP: + if effect == EFFECT_COLORLOOP and color_loop_supported: result = await self._color_cluster.color_loop_set( update_flags=( Color.ColorLoopUpdateFlags.Action @@ -657,8 +688,9 @@ async def _async_turn_on_impl( # noqa: C901 start_hue=0, ) t_log["color_loop_set"] = result - self._effect = EFFECT_COLORLOOP - elif self._effect == EFFECT_COLORLOOP and effect != EFFECT_COLORLOOP: + if result[1] is Status.SUCCESS: + self._effect = EFFECT_COLORLOOP + elif deactivate_color_loop_after_turn_on: result = await self._color_cluster.color_loop_set( update_flags=Color.ColorLoopUpdateFlags.Action, action=Color.ColorLoopAction.Deactivate, @@ -667,7 +699,8 @@ async def _async_turn_on_impl( # noqa: C901 start_hue=0, ) t_log["color_loop_set"] = result - self._effect = EFFECT_OFF + if result[1] is Status.SUCCESS: + self._effect = EFFECT_OFF if flash is not None: assert self._identify_cluster is not None From bc5acaf061cefcfbe706490cdf68b44d26565202 Mon Sep 17 00:00:00 2001 From: David Mulcahey Date: Tue, 14 Jul 2026 10:10:47 -0400 Subject: [PATCH 2/2] Narrow color loop deactivation coverage --- tests/test_light.py | 48 +++++++++------------ zha/application/platforms/light/__init__.py | 11 +---- 2 files changed, 23 insertions(+), 36 deletions(-) diff --git a/tests/test_light.py b/tests/test_light.py index ba6497551..5d50683c5 100644 --- a/tests/test_light.py +++ b/tests/test_light.py @@ -1940,8 +1940,8 @@ async def test_turn_on_colorloop_deactivation_order_and_guard( @pytest.mark.parametrize( - "status", - [zcl_f.Status.FAILURE, zcl_f.Status.UNSUP_CLUSTER_COMMAND], + "execute_if_off", + [False, True], ) @patch( "zigpy.zcl.clusters.lighting.Color.request", @@ -1957,13 +1957,17 @@ async def test_turn_on_colorloop_deactivation_order_and_guard( ) async def test_colorloop_deactivation_failure_preserves_independent_commands( zha_gateway: Gateway, - status: zcl_f.Status, + execute_if_off: bool, ) -> None: """Test deactivation failure blocks only the dependent color command.""" device_light = await device_light_3_mock(zha_gateway) entity = get_entity(device_light, platform=Platform.LIGHT) cluster_color = device_light.device.endpoints[1].light_color cluster_level = device_light.device.endpoints[1].level + cluster_color.PLUGGED_ATTR_READS = { + "options": (lighting.Color.Options.Execute_if_off if execute_if_off else 0) + } + update_attribute_cache(cluster_color) _restore_colorloop_test_state( entity, @@ -1973,7 +1977,7 @@ async def test_colorloop_deactivation_failure_preserves_independent_commands( ) cluster_color.request.reset_mock() cluster_level.request.reset_mock() - cluster_color.request.return_value = [sentinel.data, status] + cluster_color.request.return_value = [sentinel.data, zcl_f.Status.FAILURE] await entity.async_turn_on(brightness=150, color_temp=235, effect="off") @@ -1995,17 +1999,11 @@ async def test_colorloop_deactivation_failure_preserves_independent_commands( @pytest.mark.parametrize( - ("device_factory", "status", "expected_effect", "sends_command"), + ("status", "expected_effect"), [ - (device_light_3_mock, zcl_f.Status.SUCCESS, "colorloop", True), - (device_light_3_mock, zcl_f.Status.FAILURE, "off", True), - ( - device_light_3_mock, - zcl_f.Status.UNSUP_CLUSTER_COMMAND, - "off", - True, - ), - (device_light_1_mock, zcl_f.Status.SUCCESS, "off", False), + (zcl_f.Status.SUCCESS, "colorloop"), + (zcl_f.Status.FAILURE, "off"), + (zcl_f.Status.UNSUP_CLUSTER_COMMAND, "off"), ], ) @patch( @@ -2018,20 +2016,18 @@ async def test_colorloop_deactivation_failure_preserves_independent_commands( ) async def test_requested_colorloop_activation_policy( zha_gateway: Gateway, - device_factory: Any, status: zcl_f.Status, expected_effect: str, - sends_command: bool, ) -> None: - """Test activation requires support and a successful command response.""" - device_light = await device_factory(zha_gateway) + """Test activation updates the effect only after a successful response.""" + device_light = await device_light_3_mock(zha_gateway) entity = get_entity(device_light, platform=Platform.LIGHT) cluster_color = device_light.device.endpoints[1].light_color _restore_colorloop_test_state( entity, state=True, - effect="off" if sends_command else "colorloop", + effect="off", ) assert entity.state["effect"] == "off" cluster_color.request.reset_mock() @@ -2039,15 +2035,13 @@ async def test_requested_colorloop_activation_policy( await entity.async_turn_on(effect="colorloop") - expected_commands = ( - [cluster_color.commands_by_name["color_loop_set"].id] if sends_command else [] + assert _color_command_ids(cluster_color) == [ + cluster_color.commands_by_name["color_loop_set"].id + ] + assert ( + cluster_color.request.await_args.kwargs["action"] + is lighting.Color.ColorLoopAction.Activate_from_current_hue ) - assert _color_command_ids(cluster_color) == expected_commands - if sends_command: - assert ( - cluster_color.request.await_args.kwargs["action"] - is lighting.Color.ColorLoopAction.Activate_from_current_hue - ) assert entity.state["effect"] == expected_effect diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index 93c2ad8aa..4a4305f2a 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -256,12 +256,7 @@ def restore_external_state_attributes( if color_mode is not None and color_mode in self._supported_color_modes: self._color_mode = color_mode if effect is not None: - self._effect = ( - EFFECT_OFF - if effect == EFFECT_COLORLOOP - and EFFECT_COLORLOOP not in (self._effect_list or ()) - else effect - ) + self._effect = effect class BaseSharedLight(BaseLight): @@ -524,8 +519,6 @@ async def _async_turn_on_impl( # noqa: C901 t_log = {} color_loop_supported = EFFECT_COLORLOOP in (self._effect_list or ()) - if not color_loop_supported and self._effect == EFFECT_COLORLOOP: - self._effect = EFFECT_OFF color_loop_remembered = ( color_loop_supported and self._effect == EFFECT_COLORLOOP ) @@ -675,7 +668,7 @@ async def _async_turn_on_impl( # noqa: C901 self.async_transition_start_timer(transition_time) if self._color_cluster is not None: - if effect == EFFECT_COLORLOOP and color_loop_supported: + if effect == EFFECT_COLORLOOP: result = await self._color_cluster.color_loop_set( update_flags=( Color.ColorLoopUpdateFlags.Action