From 14cd399f70def13a7a3f1412304c54aaa264352e Mon Sep 17 00:00:00 2001 From: jkingsman Date: Fri, 26 Jun 2026 21:06:14 -0700 Subject: [PATCH] Suppress expected reconnect. Closes #305. --- app/fanout/mqtt_base.py | 19 ++++++++++-- tests/test_mqtt.py | 68 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/app/fanout/mqtt_base.py b/app/fanout/mqtt_base.py index 61ec2dd..9f6fb40 100644 --- a/app/fanout/mqtt_base.py +++ b/app/fanout/mqtt_base.py @@ -66,6 +66,11 @@ class BaseMqttPublisher(ABC): self.integration_name: str = "" self._last_error: str | None = None self._error_notified: bool = False + # Set when the loop tears down a healthy connection for a planned, + # self-healing reconnect (e.g. community-MQTT JWT renewal). The next + # successful connect then skips the "connected" success toast so these + # expected reconnects don't spam notifications. See issue #305. + self._suppress_next_connect_toast: bool = False def set_integration_name(self, name: str) -> None: """Attach the configured fanout-module name for operator-facing logs.""" @@ -106,6 +111,7 @@ class BaseMqttPublisher(ABC): self.connected = False self._last_error = None self._error_notified = False + self._suppress_next_connect_toast = False async def restart(self, settings: object) -> None: """Called when settings change — stop + start.""" @@ -222,8 +228,13 @@ class BaseMqttPublisher(ABC): self._error_notified = False backoff = _BACKOFF_MIN - title, detail = self._on_connected(settings) - broadcast_success(title, detail) + if self._suppress_next_connect_toast: + # Planned/self-healing reconnect — health still updates, + # but skip the redundant "connected" toast. + self._suppress_next_connect_toast = False + else: + title, detail = self._on_connected(settings) + broadcast_success(title, detail) await self._on_connected_async(settings) _broadcast_health() @@ -238,6 +249,10 @@ class BaseMqttPublisher(ABC): elapsed = time.monotonic() - connect_time await self._on_periodic_wake(elapsed) if self._should_break_wait(elapsed): + # Expected, self-healing reconnect surfaced by the + # periodic wake (e.g. community-MQTT JWT renewal), + # so don't toast "connected" when it comes back. + self._suppress_next_connect_toast = True break continue diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py index 1da4365..09bc7f8 100644 --- a/tests/test_mqtt.py +++ b/tests/test_mqtt.py @@ -235,6 +235,74 @@ class TestConnectionLoop: await pub.stop() assert pub.connected is False + @pytest.mark.asyncio + async def test_planned_reconnect_suppresses_success_toast(self): + """A self-healing reconnect (via _should_break_wait) must not re-toast. + + Regression for #305: the community-MQTT JWT-renewal teardown reconnects + on the periodic wake. The first connect should toast, but the planned + reconnect should only refresh health, not pop another success toast. + """ + import asyncio + + pub = MqttPublisher() + settings = _make_settings() + + real_wait_for = asyncio.wait_for + connect_count = 0 + second_connect = asyncio.Event() + break_done = {"value": False} + + def factory(**kwargs): + mock = _mock_aiomqtt_client() + original_aenter = mock.__aenter__ + + async def signal_aenter(*a, **kw): + nonlocal connect_count + result = await original_aenter(*a, **kw) + connect_count += 1 + if connect_count >= 2: + second_connect.set() + return result + + mock.__aenter__ = AsyncMock(side_effect=signal_aenter) + return mock + + async def fake_wait_for(coro, timeout): + # Fire the ~60s housekeeping wake immediately; delegate every other + # wait (including the test's own) to the real implementation. + if timeout == 60: + if asyncio.iscoroutine(coro): + coro.close() + await asyncio.sleep(0) # yield so the test coroutine can run + raise TimeoutError + return await real_wait_for(coro, timeout) + + def should_break(self, elapsed): + # Break exactly once to model a single planned reconnect. + if not break_done["value"]: + break_done["value"] = True + return True + return False + + with ( + patch("app.fanout.mqtt_base.aiomqtt.Client", side_effect=factory), + patch("app.fanout.mqtt_base._broadcast_health"), + patch("app.fanout.mqtt_base.asyncio.wait_for", side_effect=fake_wait_for), + patch.object(MqttPublisher, "_should_break_wait", should_break), + patch("app.websocket.broadcast_success") as mock_success, + patch("app.websocket.broadcast_health"), + ): + await pub.start(settings) + await asyncio.wait_for(second_connect.wait(), timeout=5) + await pub.stop() + + assert connect_count >= 2 + # First connect toasts; the planned reconnect does not. + assert mock_success.call_count == 1 + # Flag is consumed, not left armed for a future genuine (re)connect. + assert pub._suppress_next_connect_toast is False + @pytest.mark.asyncio async def test_reconnects_after_connection_failure(self): """Connection loop should retry after a connection error with backoff."""