Merge pull request #34 from aaearon/fix/flaky-observe-resubscribe

test: remove timing race from coordinator observe-mode tests

tests/localthings/conftest.py

@@ -3,8 +3,10 @@ from __future__ import annotations
 
 import json
 from pathlib import Path
+from typing import Any
 from unittest.mock import patch
 
+import cbor2
 import pytest
 from pytest_homeassistant_custom_component.common import MockConfigEntry
 
@@ -150,12 +152,26 @@ class FakeObserveSession:
         self.subscribed: list[str] = []
         self.fail_hrefs: set[str] = set()
         self.closed = False
+        # When set to a rep dict, subscribe() immediately delivers that rep
+        # as an OBSERVE notification for the href — what a real device does
+        # when it answers a subscription with the current representation.
+        # `try_enter_observe_mode` clears its notified set before it
+        # subscribes, so this is the only way to deliver a notify that
+        # reliably counts: it lands synchronously via on_notification,
+        # after that clear and before the post-sleep fraction check — a
+        # notify raced in from another thread can be wiped by the clear
+        # (or arrive after the check) depending on scheduling. Set to
+        # None to model a device that answers subscriptions but never
+        # notifies.
+        self.notify_on_subscribe: dict[str, Any] | None = None
 
     def subscribe(self, path_segs):
         href = '/' + '/'.join(path_segs)
         if href in self.fail_hrefs:
             raise ConnectionError("subscribe failed")
         self.subscribed.append(href)
+        if self.notify_on_subscribe is not None and self.on_notification is not None:
+            self.on_notification(href, cbor2.dumps(self.notify_on_subscribe))
         return b'\x01'
 
     def refresh_observes(self, paths):

tests/localthings/test_coordinator.py

@@ -1,8 +1,6 @@
 """Tests for the LocalThingsCoordinator."""
 from __future__ import annotations
 
-import threading
-import time
 from datetime import timedelta
 from unittest.mock import AsyncMock, patch
 
@@ -260,21 +258,18 @@ async def test_enters_observe_mode_when_hot_warm_hrefs_notify(
     coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
 
-    # try_enter_observe_mode clears prior notifications as soon as it
-    # starts, so notifies must land *during* its grace-period sleep (same
-    # pattern test_observe.py uses), not before the call.
-    def _notify_during_grace_period():
-        time.sleep(0.005)
-        for href in hrefs:
-            fake.on_notification(href, cbor2.dumps({'notified': True}))
-
-    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
-    notifier.start()
+    # try_enter_observe_mode clears prior notifications before it
+    # subscribes, then checks the notified fraction after the grace-period
+    # sleep — a notify counts as long as it arrives in that window, not
+    # specifically during the sleep itself. The fake session delivers one
+    # synchronously from subscribe() for every href it subscribes to (see
+    # FakeObserveSession.notify_on_subscribe in conftest.py), which lands
+    # it in that window by construction rather than by timing.
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
 
     assert entered is True
     assert coordinator._observe.mode == MODE_OBSERVE
@@ -300,18 +295,11 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll(
 
     # Get the coordinator into observe mode the same way
     # test_enters_observe_mode_when_hot_warm_hrefs_notify does.
-    def _notify_during_grace_period():
-        time.sleep(0.005)
-        for href in hrefs:
-            fake.on_notification(href, cbor2.dumps({'notified': True}))
-
-    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
-    notifier.start()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -324,12 +312,12 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll(
 
     # Simulate the existing "poll failed, reconnecting" branch: _poll_once
     # fails once (triggering the reconnect/backoff path), then succeeds.
-    # No fresh notifies are supplied for the immediate resubscribe attempt
-    # this now triggers, so its grace period (shortened for tests by the
-    # `_fast_coordinator_timers` autouse fixture — see conftest.py) times
-    # out and mode stays 'poll' — this test only asserts the tear-down half of
-    # the fix; test_reconnect_from_observe_mode_resubscribes_immediately
-    # covers the successful-immediate-resubscribe half.
+    # The device stops notifying on subscribe, so the immediate resubscribe
+    # attempt this now triggers gets nothing back and mode stays 'poll' —
+    # this test only asserts the tear-down half of the fix;
+    # test_reconnect_from_observe_mode_resubscribes_immediately covers the
+    # successful-immediate-resubscribe half.
+    fake.notify_on_subscribe = None
     with (
         patch(
             'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
@@ -362,18 +350,11 @@ async def test_poll_timeout_skips_reconnect_when_push_is_healthy(
     coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
 
-    def _notify_during_grace_period():
-        time.sleep(0.005)
-        for href in hrefs:
-            fake.on_notification(href, cbor2.dumps({'notified': True}))
-
-    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
-    notifier.start()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -408,18 +389,11 @@ async def test_poll_timeout_reconnects_after_consecutive_limit_even_with_push(
     coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
 
-    def _notify_during_grace_period():
-        time.sleep(0.005)
-        for href in hrefs:
-            fake.on_notification(href, cbor2.dumps({'notified': True}))
-
-    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
-    notifier.start()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -441,7 +415,10 @@ async def test_poll_timeout_reconnects_after_consecutive_limit_even_with_push(
             assert coordinator.observe_mode == MODE_OBSERVE
 
     # The final consecutive timeout crosses the limit and triggers the
-    # existing reconnect path, which then succeeds and downgrades.
+    # existing reconnect path, which then succeeds and downgrades. The
+    # device stops notifying on subscribe, so the resubscribe attempt that
+    # follows the reconnect finds nothing and mode stays 'poll'.
+    fake.notify_on_subscribe = None
     with (
         patch(
             'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
@@ -473,18 +450,11 @@ async def test_poll_timeout_counter_resets_when_push_is_healthy_again(
     coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
 
-    def _notify_during_grace_period():
-        time.sleep(0.005)
-        for href in hrefs:
-            fake.on_notification(href, cbor2.dumps({'notified': True}))
-
-    notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
-    notifier.start()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -526,21 +496,11 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately(
     coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
 
-    def _notify(delay: float = 0.005) -> threading.Thread:
-        def _run():
-            time.sleep(delay)
-            for href in hrefs:
-                fake.on_notification(href, cbor2.dumps({'notified': True}))
-        t = threading.Thread(target=_run, daemon=True)
-        t.start()
-        return t
-
-    notifier = _notify()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -563,10 +523,8 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately(
             new=AsyncMock(),
         ),
     ):
-        notifier = _notify()
         await coordinator.async_request_refresh()
         await hass.async_block_till_done()
-        notifier.join()
 
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -587,21 +545,11 @@ async def test_sweep_mismatch_never_downgrades_a_live_observe_session(
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
     assert len(hrefs) >= 2
 
-    def _notify(delay: float = 0.005) -> threading.Thread:
-        def _run():
-            time.sleep(delay)
-            for href in hrefs:
-                fake.on_notification(href, cbor2.dumps({'notified': True}))
-        t = threading.Thread(target=_run, daemon=True)
-        t.start()
-        return t
-
-    notifier = _notify()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE
 
@@ -637,21 +585,11 @@ async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session(
     hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
     assert len(hrefs) >= 2
 
-    def _notify(delay: float = 0.005) -> threading.Thread:
-        def _run():
-            time.sleep(delay)
-            for href in hrefs:
-                fake.on_notification(href, cbor2.dumps({'notified': True}))
-        t = threading.Thread(target=_run, daemon=True)
-        t.start()
-        return t
-
-    notifier = _notify()
+    fake.notify_on_subscribe = {'notified': True}
     entered = await hass.async_add_executor_job(
         coordinator._observe.try_enter_observe_mode,
         fake, hrefs, 0.02, 0.8,
     )
-    notifier.join()
     assert entered is True
     assert coordinator.observe_mode == MODE_OBSERVE