Fix listener leak in HomeKit on reload (#56143)

* Fix listener leak in HomeKit on reload

* Fix mocking

homeassistant/components/homekit/__init__.py

@@ -554,7 +554,7 @@ class HomeKit:
         acc = self.driver.accessory
         if acc.entity_id not in entity_ids:
             return
-        acc.async_stop()
+        await acc.stop()
         if not (state := self.hass.states.get(acc.entity_id)):
             _LOGGER.warning(
                 "The underlying entity %s disappeared during reset", acc.entity
@@ -577,7 +577,7 @@ class HomeKit:
                 self._name,
                 entity_id,
             )
-            acc = self.remove_bridge_accessory(aid)
+            acc = await self.async_remove_bridge_accessory(aid)
             if state := self.hass.states.get(acc.entity_id):
                 new.append(state)
             else:
@@ -670,11 +670,11 @@ class HomeKit:
             )
         )
 
-    def remove_bridge_accessory(self, aid):
+    async def async_remove_bridge_accessory(self, aid):
         """Try adding accessory to bridge if configured beforehand."""
         acc = self.bridge.accessories.pop(aid, None)
         if acc:
-            acc.async_stop()
+            await acc.stop()
         return acc
 
     async def async_configure_accessories(self):
@@ -866,11 +866,6 @@ class HomeKit:
         self.status = STATUS_STOPPED
         _LOGGER.debug("Driver stop for %s", self._name)
         await self.driver.async_stop()
-        if self.bridge:
-            for acc in self.bridge.accessories.values():
-                acc.async_stop()
-        else:
-            self.driver.accessory.async_stop()
 
     @callback
     def _async_configure_linked_sensors(self, ent_reg_ent, device_lookup, state):

homeassistant/components/homekit/accessories.py

@@ -499,8 +499,7 @@ class HomeAccessory(Accessory):
             )
         )
 
-    @ha_callback
+    async def stop(self):
-    def async_stop(self):
         """Cancel any subscriptions when the bridge is stopped."""
         while self._subscriptions:
             self._subscriptions.pop(0)()

homeassistant/components/homekit/type_cameras.py

@@ -244,17 +244,21 @@ class Camera(HomeAccessory, PyhapCamera):
         Run inside the Home Assistant event loop.
         """
         if self._char_motion_detected:
-            async_track_state_change_event(
+            self._subscriptions.append(
-                self.hass,
+                async_track_state_change_event(
-                [self.linked_motion_sensor],
+                    self.hass,
-                self._async_update_motion_state_event,
+                    [self.linked_motion_sensor],
+                    self._async_update_motion_state_event,
+                )
             )
 
         if self._char_doorbell_detected:
-            async_track_state_change_event(
+            self._subscriptions.append(
-                self.hass,
+                async_track_state_change_event(
-                [self.linked_doorbell_sensor],
+                    self.hass,
-                self._async_update_doorbell_state_event,
+                    [self.linked_doorbell_sensor],
+                    self._async_update_doorbell_state_event,
+                )
             )
 
         await super().run()
@@ -434,6 +438,12 @@ class Camera(HomeAccessory, PyhapCamera):
         self.sessions[session_id].pop(FFMPEG_WATCHER)()
         self.sessions[session_id].pop(FFMPEG_LOGGER).cancel()
 
+    async def stop(self):
+        """Stop any streams when the accessory is stopped."""
+        for session_info in self.sessions.values():
+            asyncio.create_task(self.stop_stream(session_info))
+        await super().stop()
+
     async def stop_stream(self, session_info):
         """Stop the stream for the given ``session_id``."""
         session_id = session_info["id"]

homeassistant/components/homekit/type_covers.py

@@ -122,10 +122,12 @@ class GarageDoorOpener(HomeAccessory):
         Run inside the Home Assistant event loop.
         """
         if self.linked_obstruction_sensor:
-            async_track_state_change_event(
+            self._subscriptions.append(
-                self.hass,
+                async_track_state_change_event(
-                [self.linked_obstruction_sensor],
+                    self.hass,
-                self._async_update_obstruction_event,
+                    [self.linked_obstruction_sensor],
+                    self._async_update_obstruction_event,
+                )
             )
 
         await super().run()

homeassistant/components/homekit/type_humidifiers.py

@@ -149,10 +149,12 @@ class HumidifierDehumidifier(HomeAccessory):
         Run inside the Home Assistant event loop.
         """
         if self.linked_humidity_sensor:
-            async_track_state_change_event(
+            self._subscriptions.append(
-                self.hass,
+                async_track_state_change_event(
-                [self.linked_humidity_sensor],
+                    self.hass,
-                self.async_update_current_humidity_event,
+                    [self.linked_humidity_sensor],
+                    self.async_update_current_humidity_event,
+                )
             )
 
         await super().run()

tests/components/homekit/test_accessories.py

@@ -60,7 +60,7 @@ async def test_accessory_cancels_track_state_change_on_stop(hass, hk_driver):
     ):
         await acc.run()
     assert len(hass.data[TRACK_STATE_CHANGE_CALLBACKS][entity_id]) == 1
-    acc.async_stop()
+    await acc.stop()
     assert entity_id not in hass.data[TRACK_STATE_CHANGE_CALLBACKS]
 
 

tests/components/homekit/test_homekit.py

@@ -442,11 +442,12 @@ async def test_homekit_remove_accessory(hass, mock_zeroconf):
     homekit.driver = "driver"
     homekit.bridge = _mock_pyhap_bridge()
     acc_mock = MagicMock()
+    acc_mock.stop = AsyncMock()
     homekit.bridge.accessories = {6: acc_mock}
 
-    acc = homekit.remove_bridge_accessory(6)
+    acc = await homekit.async_remove_bridge_accessory(6)
     assert acc is acc_mock
-    assert acc_mock.async_stop.called
+    assert acc_mock.stop.called
     assert len(homekit.bridge.accessories) == 0
 
 
@@ -695,9 +696,11 @@ async def test_homekit_reset_accessories(hass, mock_zeroconf):
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -730,9 +733,12 @@ async def test_homekit_unpair(hass, device_reg, mock_zeroconf):
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         state = homekit.driver.state
         state.add_paired_client("client1", "any", b"1")
@@ -769,9 +775,12 @@ async def test_homekit_unpair_missing_device_id(hass, device_reg, mock_zeroconf)
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         state = homekit.driver.state
         state.add_paired_client("client1", "any", b"1")
@@ -807,6 +816,8 @@ async def test_homekit_unpair_not_homekit_device(hass, device_reg, mock_zeroconf
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
@@ -856,9 +867,12 @@ async def test_homekit_reset_accessories_not_supported(hass, mock_zeroconf):
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -896,9 +910,12 @@ async def test_homekit_reset_accessories_state_missing(hass, mock_zeroconf):
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -935,9 +952,12 @@ async def test_homekit_reset_accessories_not_bridged(hass, mock_zeroconf):
 
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         aid = homekit.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
         homekit.bridge.accessories = {aid: acc_mock}
         homekit.status = STATUS_RUNNING
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -974,7 +994,10 @@ async def test_homekit_reset_single_accessory(hass, mock_zeroconf):
         homekit.status = STATUS_RUNNING
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         homekit.driver.accessory = acc_mock
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -1008,7 +1031,10 @@ async def test_homekit_reset_single_accessory_unsupported(hass, mock_zeroconf):
         homekit.status = STATUS_RUNNING
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         homekit.driver.accessory = acc_mock
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -1041,7 +1067,10 @@ async def test_homekit_reset_single_accessory_state_missing(hass, mock_zeroconf)
         homekit.status = STATUS_RUNNING
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         homekit.driver.accessory = acc_mock
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,
@@ -1074,7 +1103,10 @@ async def test_homekit_reset_single_accessory_no_match(hass, mock_zeroconf):
         homekit.status = STATUS_RUNNING
         acc_mock = MagicMock()
         acc_mock.entity_id = entity_id
+        acc_mock.stop = AsyncMock()
+
         homekit.driver.accessory = acc_mock
+        homekit.driver.aio_stop_event = MagicMock()
 
         await hass.services.async_call(
             DOMAIN,

tests/components/homekit/test_type_cameras.py

@@ -1,5 +1,6 @@
 """Test different accessory types: Camera."""
 
+import asyncio
 from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
 from uuid import UUID
 
@@ -45,6 +46,7 @@ PID_THAT_WILL_NEVER_BE_ALIVE = 2147483647
 async def _async_start_streaming(hass, acc):
     """Start streaming a camera."""
     acc.set_selected_stream_configuration(MOCK_START_STREAM_TLV)
+    await hass.async_block_till_done()
     await acc.run()
     await hass.async_block_till_done()
 
@@ -92,6 +94,18 @@ def run_driver(hass):
         )
 
 
+def _mock_reader():
+    """Mock ffmpeg reader."""
+
+    async def _readline(*args, **kwargs):
+        await asyncio.sleep(0.1)
+
+    async def _get_reader(*args, **kwargs):
+        return AsyncMock(readline=_readline)
+
+    return _get_reader
+
+
 def _get_exits_after_startup_mock_ffmpeg():
     """Return a ffmpeg that will have an invalid pid."""
     ffmpeg = MagicMock()
@@ -99,7 +113,7 @@ def _get_exits_after_startup_mock_ffmpeg():
     ffmpeg.open = AsyncMock(return_value=True)
     ffmpeg.close = AsyncMock(return_value=True)
     ffmpeg.kill = AsyncMock(return_value=True)
-    ffmpeg.get_reader = AsyncMock()
+    ffmpeg.get_reader = _mock_reader()
     return ffmpeg
 
 
@@ -109,7 +123,7 @@ def _get_working_mock_ffmpeg():
     ffmpeg.open = AsyncMock(return_value=True)
     ffmpeg.close = AsyncMock(return_value=True)
     ffmpeg.kill = AsyncMock(return_value=True)
-    ffmpeg.get_reader = AsyncMock()
+    ffmpeg.get_reader = _mock_reader()
     return ffmpeg
 
 
@@ -120,7 +134,7 @@ def _get_failing_mock_ffmpeg():
     ffmpeg.open = AsyncMock(return_value=False)
     ffmpeg.close = AsyncMock(side_effect=OSError)
     ffmpeg.kill = AsyncMock(side_effect=OSError)
-    ffmpeg.get_reader = AsyncMock()
+    ffmpeg.get_reader = _mock_reader()
     return ffmpeg