Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions tests/test_light.py
Original file line number Diff line number Diff line change
Expand Up @@ -2069,6 +2069,61 @@ async def test_turn_on_during_off_transition(zha_gateway: Gateway) -> None:
assert bool(entity.state["on"]) is False


@patch(
"zigpy.zcl.clusters.general.OnOff.request",
new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]),
)
async def test_on_off_only_rapid_toggle_does_not_revert(
zha_gateway: Gateway,
) -> None:
"""Test rapid turn_off then turn_on on an on/off-only light does not revert to off.

On/off-only lights have no brightness transition to wait for, so turn_off
must not arm the transitioning flag/timer. Otherwise a buffered on_off=False
report from the off command would override the optimistic on-state when
the timer fires after the follow-up turn_on.
"""
zigpy_device = create_mock_zigpy_device(zha_gateway, LIGHT_ON_OFF)
on_off_cluster = zigpy_device.endpoints[1].on_off
on_off_cluster.PLUGGED_ATTR_READS = {"on_off": 1}
zha_device = await join_zigpy_device(zha_gateway, zigpy_device)

entity = get_entity(zha_device, platform=Platform.LIGHT)
assert bool(entity.state["on"]) is True

# Turn off, then immediately turn back on (within the default ~1.5s window).
await entity.async_turn_off()
await zha_gateway.async_block_till_done()
assert bool(entity.state["on"]) is False
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: is bool(...) needed? The data type of on should be bool.

# On/off-only lights have no transition to wait for.
assert not entity.is_transitioning

# The device's on_off=False report from the off command arrives.
await send_attributes_report(
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if it's useful to have as an explicit test but this also passes (as expected) when the reports both come in after the commands succeed (i.e. if you manage to spam the on/off toggle faster than the light reports).

zha_gateway,
on_off_cluster,
{general.OnOff.AttributeDefs.on_off.id: 0},
)
await zha_gateway.async_block_till_done()

await entity.async_turn_on()
await zha_gateway.async_block_till_done()
assert bool(entity.state["on"]) is True

# The device's on_off=True report from the on command arrives.
await send_attributes_report(
zha_gateway,
on_off_cluster,
{general.OnOff.AttributeDefs.on_off.id: 1},
)
await zha_gateway.async_block_till_done()

# Wait past the old 1.5s timer window. State must remain on.
await asyncio.sleep(2)
await zha_gateway.async_block_till_done()
Comment on lines +2121 to +2123
Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We already have this elsewhere and this isn't a real 2 second wait (due to pytest-looptime). I think this is fine.

assert bool(entity.state["on"]) is True


async def test_light_state_restoration(zha_gateway: Gateway) -> None:
"""Test the light state restoration function."""
device_light_3 = await device_light_3_mock(zha_gateway)
Expand Down
15 changes: 9 additions & 6 deletions zha/application/platforms/light/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -633,8 +633,13 @@ async def async_turn_off(self, *, transition: float | None = None) -> None:
else DEFAULT_ON_OFF_TRANSITION
) + DEFAULT_EXTRA_TRANSITION_DELAY_SHORT

# Start pausing attribute report parsing
if self._zha_config_enable_light_transitioning_flag:
# Start pausing attribute report parsing, but not for on/off-only
# lights (they have no transition to wait for, and a follow-up
# turn_on would not clear the buffered off-report).
set_transition_flag = (
brightness_supported and self._zha_config_enable_light_transitioning_flag
)
if set_transition_flag:
self.async_transition_set_flag()

try:
Expand All @@ -654,7 +659,7 @@ async def async_turn_off(self, *, transition: float | None = None) -> None:
result = await self._on_off_cluster_handler.off()

# Pause parsing attribute reports until transition is complete
if self._zha_config_enable_light_transitioning_flag:
if set_transition_flag:
self.async_transition_start_timer(transition_time)
self.debug("turned off: %s", result)
if result[1] is not Status.SUCCESS:
Expand All @@ -676,9 +681,7 @@ async def async_turn_off(self, *, transition: float | None = None) -> None:
finally:
# If the task was cancelled before the transition timer was started,
# clean up the transitioning flag so the light does not get stuck.
self._async_cleanup_transition_if_stuck(
self._zha_config_enable_light_transitioning_flag
)
self._async_cleanup_transition_if_stuck(set_transition_flag)

async def async_handle_color_commands(
self,
Expand Down
Loading