Merge pull request #391 from JayChickenK/observe-silent-href-subpolls

Keep asking the appliance for readings that never send live updates

tests/localthings/test_coordinator.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
 import time
 from datetime import timedelta
 from unittest.mock import AsyncMock, patch
@@ -1050,6 +1052,110 @@ async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session(
     mock_subpolls.assert_called_once_with(force=True)
 
 
+async def _cancel_background_subpolls(coordinator: LocalThingsCoordinator) -> None:
+    """Setup starts `_run_subpolls` as a background task. Tests that drive
+    it directly have to cancel that one first, or a patched
+    `_poll_hrefs_blocking` also captures its batches."""
+    task = coordinator._subpoll_task
+    if task is None:
+        return
+    task.cancel()
+    coordinator._subpoll_task = None
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+
+async def test_observe_mode_subpolls_only_silent_hrefs(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """Issue #92: subscribed-but-silent hrefs keep the hot/warm cadence
+    instead of waiting for the 30s sweep."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    await _cancel_background_subpolls(coordinator)
+    assert coordinator._hot_hrefs
+    silent = coordinator._hot_hrefs[0]
+    coordinator._observe.mode = MODE_OBSERVE
+    coordinator._observe.fallback_hrefs = {silent}
+
+    polled: list[list[str]] = []
+
+    def _capture(hrefs):
+        polled.append(list(hrefs))
+
+    with (
+        patch.object(coordinator, "_poll_hrefs_blocking", side_effect=_capture),
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new_callable=AsyncMock,
+        ),
+    ):
+        await coordinator._run_subpolls()
+
+    assert polled
+    for batch in polled:
+        assert set(batch) == {silent}
+
+
+async def test_observe_mode_skips_subpolls_when_nothing_is_silent(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """The observe-mode no-op stays in place when every subscribed href
+    actually notified -- issue #92 only keeps the silent ones on poll."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    await _cancel_background_subpolls(coordinator)
+    coordinator._observe.mode = MODE_OBSERVE
+    coordinator._observe.fallback_hrefs = set()
+
+    with (
+        patch.object(coordinator, "_poll_hrefs_blocking") as mock_poll,
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new_callable=AsyncMock,
+        ) as mock_sleep,
+    ):
+        await coordinator._run_subpolls()
+
+    mock_poll.assert_not_called()
+    mock_sleep.assert_not_called()
+
+
+async def test_observe_mode_skips_empty_subpoll_slots(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
+) -> None:
+    """Once hot is empty, odd slots would otherwise take the session lock
+    and dispatch a no-op executor job. Skip those."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    await _cancel_background_subpolls(coordinator)
+    coordinator._observe.mode = MODE_OBSERVE
+    coordinator._hot_hrefs = []
+    coordinator._warm_hrefs = ["/warm/vs/0"]
+    coordinator._observe.fallback_hrefs = {"/warm/vs/0"}
+
+    polled: list[list[str]] = []
+
+    def _capture(hrefs):
+        polled.append(list(hrefs))
+
+    with (
+        patch.object(coordinator, "_poll_hrefs_blocking", side_effect=_capture),
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new_callable=AsyncMock,
+        ),
+    ):
+        await coordinator._run_subpolls()
+
+    assert polled
+    for batch in polled:
+        assert batch == ["/warm/vs/0"]
+
+
 async def test_write_marks_href_pending_before_post(
     hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
 ) -> None:

tests/localthings/test_observe.py

@@ -190,6 +190,7 @@ def test_try_enter_observe_mode_succeeds_when_all_hrefs_notify():
         assert entered is True
         assert mgr.mode == "observe"
         assert mgr.subscribed_hrefs == set(hrefs)
+        assert mgr.fallback_hrefs == set()
     finally:
         mgr.close()
 
@@ -218,6 +219,41 @@ def test_try_enter_observe_mode_falls_back_when_subscribe_fails_for_all():
     assert mgr.mode == "poll"
 
 
+def test_enter_observe_mode_keeps_silent_hrefs_on_fallback():
+    """Issue #92: subscribed hrefs that never notified stay on the poll
+    cadence via fallback_hrefs, rather than being treated as push-covered."""
+    mgr = _manager()
+    session = _FakeSession()
+    subscribed = {"/a/vs/0", "/b/vs/0", "/c/vs/0"}
+    mgr.on_notification("/a/vs/0", cbor2.dumps({"x": 1}))
+    mgr.on_notification("/b/vs/0", cbor2.dumps({"x": 1}))
+    try:
+        mgr.enter_observe_mode(session, subscribed)
+        assert mgr.mode == "observe"
+        assert mgr.subscribed_hrefs == subscribed
+        assert mgr.fallback_hrefs == {"/c/vs/0"}
+    finally:
+        mgr.close()
+
+
+def test_on_notification_drops_href_from_fallback():
+    """A late first notify (after the 80% snapshot) self-corrects
+    fallback_hrefs so a slow-but-pushing href is not polled for the rest
+    of the session -- multi-block resources being the reliable victim."""
+    mgr = _manager()
+    session = _FakeSession()
+    subscribed = {"/a/vs/0", "/b/vs/0", "/c/vs/0"}
+    mgr.on_notification("/a/vs/0", cbor2.dumps({"x": 1}))
+    mgr.on_notification("/b/vs/0", cbor2.dumps({"x": 1}))
+    try:
+        mgr.enter_observe_mode(session, subscribed)
+        assert mgr.fallback_hrefs == {"/c/vs/0"}
+        mgr.on_notification("/c/vs/0", cbor2.dumps({"x": 1}))
+        assert mgr.fallback_hrefs == set()
+    finally:
+        mgr.close()
+
+
 def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies():
     mgr = _manager()
     session = _FakeSession()
@@ -243,6 +279,7 @@ def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies():
 
         assert entered is True
         assert mgr.mode == "observe"
+        assert mgr.fallback_hrefs == {hrefs[3]}
     finally:
         mgr.close()