Add async_schedule_call to the Debouncer (#111051)

tests/helpers/test_debounce.py

@@ -2,7 +2,7 @@
 import asyncio
 from datetime import timedelta
 import logging
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 
@@ -69,6 +69,106 @@ async def test_immediate_works(hass: HomeAssistant) -> None:
     assert debouncer._job.target == debouncer.function
 
 
+async def test_immediate_works_with_schedule_call(hass: HomeAssistant) -> None:
+    """Test immediate works with scheduled calls."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=0.01,
+        immediate=True,
+        function=AsyncMock(side_effect=lambda: calls.append(None)),
+    )
+
+    # Call when nothing happening
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+
+    # Call when cooldown active setting execute at end to True
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+    assert debouncer._job.target == debouncer.function
+
+    # Canceling debounce in cooldown
+    debouncer.async_cancel()
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+
+    before_job = debouncer._job
+
+    # Call and let timer run out
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 2
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=1))
+    await hass.async_block_till_done()
+    assert len(calls) == 2
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+    assert debouncer._job == before_job
+
+    # Test calling doesn't execute/cooldown if currently executing.
+    await debouncer._execute_lock.acquire()
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 2
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+    debouncer._execute_lock.release()
+    assert debouncer._job.target == debouncer.function
+
+
+async def test_immediate_works_with_callback_function(hass: HomeAssistant) -> None:
+    """Test immediate works with callback function."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=0.01,
+        immediate=True,
+        function=callback(Mock(side_effect=lambda: calls.append(None))),
+    )
+
+    # Call when nothing happening
+    await debouncer.async_call()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+
+    debouncer.async_cancel()
+
+
+async def test_immediate_works_with_executor_function(hass: HomeAssistant) -> None:
+    """Test immediate works with executor function."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=0.01,
+        immediate=True,
+        function=Mock(side_effect=lambda: calls.append(None)),
+    )
+
+    # Call when nothing happening
+    await debouncer.async_call()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+
+    debouncer.async_cancel()
+
+
 async def test_immediate_works_with_passed_callback_function_raises(
     hass: HomeAssistant,
 ) -> None:
@@ -247,6 +347,61 @@ async def test_not_immediate_works(hass: HomeAssistant) -> None:
     assert debouncer._job.target == debouncer.function
 
 
+async def test_not_immediate_works_schedule_call(hass: HomeAssistant) -> None:
+    """Test immediate works with schedule call."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=0.01,
+        immediate=False,
+        function=AsyncMock(side_effect=lambda: calls.append(None)),
+    )
+
+    # Call when nothing happening
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+
+    # Call while still on cooldown
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+
+    # Canceling while on cooldown
+    debouncer.async_cancel()
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+
+    # Call and let timer run out
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=1))
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is False
+    assert debouncer._job.target == debouncer.function
+
+    # Reset debouncer
+    debouncer.async_cancel()
+
+    # Test calling doesn't schedule if currently executing.
+    await debouncer._execute_lock.acquire()
+    debouncer.async_schedule_call()
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+    debouncer._execute_lock.release()
+    assert debouncer._job.target == debouncer.function
+
+
 async def test_immediate_works_with_function_swapped(hass: HomeAssistant) -> None:
     """Test immediate works and we can change out the function."""
     calls = []