From 3ac068d734755b7beb10bb600451b346440f0dd9 Mon Sep 17 00:00:00 2001 From: Lenn Date: Thu, 21 Mar 2024 22:21:36 +0100 Subject: [PATCH] Remove references to HA --- motionblindsble/device.py | 36 ++++++++++++++++++------------------ tests/test_device.py | 30 +++++++++++++++--------------- 2 files changed, 33 insertions(+), 33 deletions(-) diff --git a/motionblindsble/device.py b/motionblindsble/device.py index 5bbdfcc..adb1b4a 100644 --- a/motionblindsble/device.py +++ b/motionblindsble/device.py @@ -234,12 +234,12 @@ def _create_connection_task(self, device: MotionDevice) -> Task | Any: """Create a connection task.""" # pylint: disable=protected-access # type: ignore[call-arg] - if device._ha_create_task: + if device._create_task: _LOGGER.debug( - "(%s) Connecting using Home Assistant", + "(%s) Connecting using create_task", device.ble_device.address, ) - return device._ha_create_task(target=device.establish_connection()) + return device._create_task(target=device.establish_connection()) _LOGGER.debug("(%s) Connecting", device.ble_device.address) return get_event_loop().create_task(device.establish_connection()) @@ -316,9 +316,9 @@ class MotionDevice: _disconnect_time: float | None _disconnect_timer: TimerHandle | Callable | None - # Callbacks that are used to interface with HA - _ha_create_task: Callable[[Coroutine], Task] | None = None - _ha_call_later: Callable[[int, Coroutine], Callable] | None = None + # Custom function that are used to create and schedule tasks + _create_task: Callable[[Coroutine], Task] | None = None + _call_later: Callable[[int, Coroutine], Callable] | None = None # Callbacks _connect_status_query_time: float | None @@ -455,17 +455,17 @@ def connection_type(self) -> MotionConnectionType: """Return the connection type.""" return self._connection_type - def set_ha_create_task( - self, ha_create_task: Callable[[Coroutine], Task] + def set_create_task_factory( + self, _create_task: Callable[[Coroutine], Task] ) -> None: """Set the create_task function to use.""" - self._ha_create_task = ha_create_task + self._create_task = _create_task - def set_ha_call_later( - self, ha_call_later: Callable[[int, Coroutine], Callable] + def set_call_later_factory( + self, _call_later: Callable[[int, Coroutine], Callable] ) -> None: """Set the call_later function to use.""" - self._ha_call_later = ha_call_later + self._call_later = _call_later def cancel_disconnect_timer(self) -> None: """Cancel the disconnect timeout.""" @@ -516,14 +516,14 @@ async def _disconnect_later(_: datetime | None = None): await self.disconnect() self._disconnect_time = new_disconnect_time - if self._ha_call_later: + if self._call_later: _LOGGER.debug( "(%s) Refreshing disconnect timeout to %.2fs" - " using Home Assistant", + " using call_later", self.ble_device.address, timeout, ) - self._disconnect_timer = self._ha_call_later( + self._disconnect_timer = self._call_later( delay=timeout, action=_disconnect_later ) # type: ignore[call-arg] else: @@ -596,12 +596,12 @@ def _disconnect_callback(self, _: BleakClient) -> None: self.update_speed(None) self._current_bleak_client = None if self._permanent_connection: - if self._ha_create_task: + if self._create_task: _LOGGER.debug( - "(%s) Automatically reconnecting using Home Assistant", + "(%s) Automatically reconnecting using create_task", self.ble_device.address, ) - self._ha_create_task(target=self.connect()) # type: ignore[call-arg] + self._create_task(target=self.connect()) # type: ignore[call-arg] else: _LOGGER.debug( "(%s) Automatically reconnecting", self.ble_device.address diff --git a/tests/test_device.py b/tests/test_device.py index ac6a724..9daf82f 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -174,15 +174,15 @@ def test_create_connection_task(self) -> None: connection_queue._create_connection_task(device) # Test creation of connect task with Home Assistant - mock_ha_create_task = Mock() - device.set_ha_create_task(mock_ha_create_task) - assert device._ha_create_task is not None + mock_create_task = Mock() + device.set_create_task_factory(mock_create_task) + assert device._create_task is not None with patch( "motionblindsble.device.MotionDevice.establish_connection", AsyncMock(return_value=True), ): connection_queue._create_connection_task(device) - mock_ha_create_task.assert_called() + mock_create_task.assert_called() @patch("motionblindsble.device.ConnectionQueue._create_connection_task") @patch("motionblindsble.device.wait") @@ -437,7 +437,7 @@ async def test_refresh_disconnect_timer( device.refresh_disconnect_timer() assert device._disconnect_time == 9999000 - # Test with _ha_call_later and _disconnect_later: + # Test with _call_later and _disconnect_later: def call_later(delay: int, action): # Run immediately instead of after delay loop = asyncio.get_event_loop() @@ -446,9 +446,9 @@ def call_later(delay: int, action): else: asyncio.run(action()) - device._ha_call_later = Mock(side_effect=call_later) + device._call_later = Mock(side_effect=call_later) device.refresh_disconnect_timer() - device._ha_call_later.assert_called_once() + device._call_later.assert_called_once() await asyncio.sleep( 0 @@ -478,7 +478,7 @@ def call_later_condition(delay: int, action: Callable) -> None: asyncio.run(call_once_condition_is_met(action)) mock_disconnect.reset_mock() - device._ha_call_later = Mock(side_effect=call_later_condition) + device._call_later = Mock(side_effect=call_later_condition) # First set it to False and pass first line of refresh_disconnect_timer # and make sure that a _disconnect_later is created await device.set_permanent_connection(False) @@ -506,11 +506,11 @@ async def test_permanent_connection(self, mock_connect) -> None: assert mock_connect.call_count == 2 # Test permanent connection with Home Assistant - mock_ha_create_task = Mock() - device.set_ha_create_task(mock_ha_create_task) + mock_create_task = Mock() + device.set_create_task_factory(mock_create_task) await device.set_permanent_connection(True) device._disconnect_callback(Mock()) - assert mock_ha_create_task.call_count == 1 + assert mock_create_task.call_count == 1 class TestDevice: @@ -554,10 +554,10 @@ def test_setters(self) -> None: device = MotionDevice("00:11:22:33:44:55") mock = Mock() mock2 = Mock() - device.set_ha_create_task(mock) - device.set_ha_call_later(mock2) - assert device._ha_create_task is mock - assert device._ha_call_later is mock2 + device.set_create_task_factory(mock) + device.set_call_later_factory(mock2) + assert device._create_task is mock + assert device._call_later is mock2 @patch("motionblindsble.device.MotionCrypt.decrypt", lambda x: x) async def test_notification_callback(self) -> None: