From c0bcccf00870bf84fbcd927a7213acfa3e7615ed Mon Sep 17 00:00:00 2001 From: Stuart Pearson <1926002+stuartp44@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:25:20 +0200 Subject: [PATCH 1/2] fix(realtime): properly dispose of stale MQTT clients before reconnecting --- custom_components/minibrew/realtime.py | 34 +++++++++++---- tests/test_realtime_overlay.py | 57 ++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 7 deletions(-) diff --git a/custom_components/minibrew/realtime.py b/custom_components/minibrew/realtime.py index c35b0ac..a86b354 100644 --- a/custom_components/minibrew/realtime.py +++ b/custom_components/minibrew/realtime.py @@ -121,8 +121,26 @@ async def async_start(self): self._stopped = False await self._async_do_connect() + def _dispose_mqtt_client(self, mqtt=None): + """Disconnect an MQTT client and drop our reference to it. + + This avoids leaking stale paho sockets and threads when the broker is + unavailable or a connection attempt fails and we immediately retry. + """ + mqtt = self._mqtt if mqtt is None else mqtt + if mqtt is None: + return + try: + mqtt.disconnect() + except Exception as err: # noqa: BLE001 + _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err) + if self._mqtt is mqtt: + self._mqtt = None + async def _async_do_connect(self): """Internal: create the MQTT client and connect. Called from async_start and reconnect.""" + if self._mqtt is not None: + self._dispose_mqtt_client() try: self._mqtt = await self.hass.async_add_executor_job(self._client.create_mqtt_client) except Exception as err: # noqa: BLE001 - never break REST polling @@ -142,6 +160,7 @@ async def _async_do_connect(self): _LOGGER.info("MiniBrew realtime: MQTT connect() returned (waiting for on_connected callback)") except Exception as err: # noqa: BLE001 - never break REST polling _LOGGER.warning("MiniBrew realtime: could not connect to MQTT broker: %s", err) + self._dispose_mqtt_client(self._mqtt) self._schedule_reconnect() def async_ensure_subscribed(self, serials): @@ -184,13 +203,13 @@ async def async_stop(self): if self._reconnect_task is not None: self._reconnect_task.cancel() self._reconnect_task = None - if self._mqtt is None: - return - mqtt, self._mqtt = self._mqtt, None - try: - await self.hass.async_add_executor_job(mqtt.disconnect) - except Exception as err: # noqa: BLE001 - _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err) + if self._mqtt is not None: + mqtt = self._mqtt + self._mqtt = None + try: + await self.hass.async_add_executor_job(mqtt.disconnect) + except Exception as err: # noqa: BLE001 + _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err) self._connected = False def _schedule_reconnect(self): @@ -266,6 +285,7 @@ def _handle_disconnected(self): """Mark disconnected, refresh entity availability, and schedule reconnect (paho thread).""" self._connected = False _LOGGER.warning("MiniBrew realtime: MQTT disconnected — will retry in %s s", _RECONNECT_DELAY) + self._dispose_mqtt_client() self.hass.loop.call_soon_threadsafe(self.coordinator.async_update_listeners) self._schedule_reconnect() diff --git a/tests/test_realtime_overlay.py b/tests/test_realtime_overlay.py index e0aff62..a37f31b 100644 --- a/tests/test_realtime_overlay.py +++ b/tests/test_realtime_overlay.py @@ -184,6 +184,63 @@ def call_soon_threadsafe(self, callback): assert len(calls) == 1 + +def test_realtime_manager_disconnects_stale_client_before_reconnect(): + import asyncio + + class _FakeMqtt: + def __init__(self, name): + self.name = name + self.disconnected = 0 + self.connected = 0 + + def on_device_log(self, *args, **kwargs): + pass + + def on_connected(self, *args, **kwargs): + pass + + def on_disconnected(self, *args, **kwargs): + pass + + def on_error(self, *args, **kwargs): + pass + + def connect(self): + self.connected += 1 + + def disconnect(self): + self.disconnected += 1 + + class _FakeLoop: + def call_soon_threadsafe(self, callback): + callback() + + class _FakeHass: + def __init__(self): + self.loop = _FakeLoop() + + async def async_add_executor_job(self, func, *args): + return func(*args) + + class _FakeCoordinator: + def async_update_listeners(self): + return None + + existing = _FakeMqtt("old") + new_client = _FakeMqtt("new") + hass = _FakeHass() + coordinator = _FakeCoordinator() + client = SimpleNamespace(create_mqtt_client=lambda: new_client) + manager = realtime.MiniBrewRealtimeManager(hass, coordinator, client) + manager._mqtt = existing + + asyncio.run(manager._async_do_connect()) + + assert existing.disconnected == 1 + assert manager._mqtt is new_client + assert new_client.connected == 1 + if __name__ == "__main__": failures = 0 for name, fn in sorted(globals().items()): From aedbb36b8d4696adf80edc9ad2073dc8b401462b Mon Sep 17 00:00:00 2001 From: Stuart Pearson <1926002+stuartp44@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:32:17 +0200 Subject: [PATCH 2/2] feat(tests): add validation step to run unit tests in PR workflow --- .github/workflows/validate-pr.yml | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/.github/workflows/validate-pr.yml b/.github/workflows/validate-pr.yml index f0c1b5c..6b9400e 100644 --- a/.github/workflows/validate-pr.yml +++ b/.github/workflows/validate-pr.yml @@ -37,6 +37,27 @@ jobs: didn't match the configured pattern. Please ensure that the subject doesn't start with an uppercase character. + validate-tests: + runs-on: ubuntu-latest + name: Run tests + + steps: + - name: Checkout code + uses: actions/checkout@v7 + + - name: Setup Python + uses: actions/setup-python@v7 + with: + python-version: '3.11' + + - name: Install test dependencies + run: | + python -m pip install --upgrade pip + pip install pytest + + - name: Run unit tests + run: python -m pytest -q tests + validate-hacs: runs-on: ubuntu-latest name: Validate HACS