From 569029c1168131f0a27d00dc5a79e7c3fec8463c Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Fri, 10 Jul 2026 19:50:40 -0500 Subject: [PATCH] refactor(observe): stop tearing down live sessions on data-drift or poll hiccups Replace the breadth/silence downgrade heuristic with a simpler rule: a still-live OBSERVE session is never torn down on a sweep/cache mismatch (the 30s sweep already corrects the cache regardless of mode) -- only a proven reconnect invalidates subscriptions. A mismatch instead triggers extra hot/warm subpolls this cycle as a bounded fallback. Also stop treating a summary-poll block-level ACK timeout as session death: distinguish TimeoutError (transfer was progressing, session likely alive) from ConnectionError (session actually closed), backed by a consecutive-timeout counter so a genuinely dead channel still recovers. This was causing a flaky/slow device (e.g. a dishwasher) to flap observe<->poll every ~45s on nothing but a slow blockwise GET. Also remove the dead is_active/active_when scaffolding (never wired into the coordinator) and add DEBUG logging for each OBSERVE notify received, including its href. --- custom_components/localthings/coordinator.py | 178 +++++++-- custom_components/localthings/observe.py | 95 ++++- .../localthings/registry/__init__.py | 4 +- .../localthings/registry/adapter.py | 10 - .../registry/capabilities/operational.py | 5 - .../localthings/registry/capabilities/oven.py | 6 - .../localthings/registry/capability.py | 3 +- tests/localthings/conftest.py | 26 ++ tests/localthings/test_coordinator.py | 378 +++++++++++++++++- tests/localthings/test_observe.py | 123 +++++- tests/test_capability.py | 1 - tests/test_end_to_end_v2.py | 4 +- tests/test_operational_capability.py | 5 - 13 files changed, 745 insertions(+), 93 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index f9f9be7..50c5aed 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import logging +import threading import time from datetime import timedelta from typing import Any @@ -24,7 +25,7 @@ from .registry.discovery import discover, BoundEntity from .registry import CAPABILITIES from .registry.adapter import flatten from .registry.identity import read_identity, DeviceIdentity -from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL +from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL, GRACE_PERIOD_S from .const import ( DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, @@ -54,10 +55,30 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): device_info: DeviceInfo device_serial: str + # Class-level knobs (not instance attrs) so tests can shrink real-time + # delays via `patch.object(LocalThingsCoordinator, ...)` without + # touching the production defaults these are computed from. Production + # code always sees these two values; only tests override them. + _SUBPOLL_STEP_S: float = SUMMARY_INTERVAL_S / 10 # 3.0 s + _OBSERVE_GRACE_PERIOD_S: float = GRACE_PERIOD_S + _RECONNECT_PAUSE_S: float = 5.0 + + # A block-level ACK timeout on the summary GET doesn't prove the + # session is dead (see _poll_once) — require this many in a row + # before treating it as one. A single slow transfer on an otherwise + # fine session shouldn't tear down a working OBSERVE subscription. + _POLL_TIMEOUT_LIMIT: int = 3 + def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: + # Per-device logger (module logger scoped to this device's host) so + # every log line — including the base DataUpdateCoordinator's own + # messages and ObserveManager's — identifies which device it's + # about. A bare module-level logger is shared across every + # configured device, which makes multi-device logs ambiguous. + self._log = logging.getLogger(f"{__name__}.{entry.data[CONF_HOST]}") super().__init__( hass, - _LOGGER, + self._log, name=f"{DOMAIN}_{entry.data[CONF_HOST]}", update_interval=timedelta(seconds=SUMMARY_INTERVAL_S), ) @@ -68,7 +89,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self.bound = [] self._cache = StateCache(_NoOpDescriptor()) self._cache.set_on_change(self._on_cache_changed) - self._observe = ObserveManager(self._cache, logger=_LOGGER) + self._observe = ObserveManager(self._cache, logger=self._log) + self._push_pending = False + self._push_pending_lock = threading.Lock() self.device_serial = entry.data[CONF_HOST] # placeholder until first poll self.device_info = DeviceInfo( identifiers={(DOMAIN, entry.data[CONF_HOST])}, @@ -81,6 +104,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._warm_hrefs: list[str] = [] self.device_type_name: str | None = None self.one_ui_version: str = '' + self._consecutive_poll_timeouts = 0 self._unbound_hrefs: list[str] = [] # ------------------------------------------------------------------ @@ -106,11 +130,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): sess.connect() sess.start_reader() self._session = sess - _LOGGER.debug("DTLS connected to %s:%d", host, port) + self._log.debug("DTLS connected to %s:%d", host, port) try: self._identity = read_identity(sess, None) except Exception as e: - _LOGGER.debug("read_identity failed: %s", e) + self._log.debug("read_identity failed: %s", e) self._identity = None def _close_session(self) -> None: @@ -133,23 +157,48 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): """StateCache.set_on_change callback. Runs on whatever thread applied the update (DTLS reader thread for observe notifications, an executor thread for poll/sweep) — never the event loop, so the - HA push must be scheduled thread-safely.""" + HA push must be scheduled thread-safely. A sweep/poll cycle can + call apply_rep for dozens of hrefs in a tight loop; coalesce those + into a single push instead of one hass.add_job per href.""" if not changed: return + with self._push_pending_lock: + if self._push_pending: + return + self._push_pending = True self.hass.add_job(self._push_cache_snapshot) @callback def _push_cache_snapshot(self) -> None: + with self._push_pending_lock: + self._push_pending = False if self.bound: self.async_set_updated_data(flatten(self.bound, self._cache.snapshot())) def _poll_once(self) -> dict[str, dict]: - """GET /device/0, return parsed resources. Blocking.""" + """GET /device/0, return parsed resources. Blocking. + + `sess.get()` raises `TimeoutError` when one block's ACK doesn't + arrive in time — the transfer was progressing (earlier blocks + succeeded) and just didn't finish before the deadline on a slow + device. That does NOT prove the session is dead, so it's left + open here; `_async_update_data` decides whether repeated timeouts + (or a lack of them) warrant a reconnect. Anything else (a + `ConnectionError` from an explicitly closed/broken session, a bad + response code) is unambiguous — close immediately. + """ if self._session is None: self._connect_session() sess = self._session try: - code, payload = sess.get(_SEED_PATH, timeout=15.0) + # A slow device can still be mid-transfer (block 8, block 11) + # when a tighter deadline cuts it off — that's a poll that + # would have succeeded, not a dead session. 35s gives a slow + # blockwise transfer room to actually finish instead of + # generating a TimeoutError every cycle. + code, payload = sess.get(_SEED_PATH, timeout=35.0) + except TimeoutError: + raise except Exception as e: self._close_session() raise RuntimeError(f"poll GET failed: {e}") from e @@ -181,24 +230,29 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._observe.apply(href, rep, source='poll') results[href] = rep except Exception as e: - _LOGGER.debug("sub-poll %s: %s", href, e) + self._log.debug("sub-poll %s: %s", href, e) return results # ------------------------------------------------------------------ # Sub-poll loop (runs between summary polls) # ------------------------------------------------------------------ - async def _run_subpolls(self) -> None: + async def _run_subpolls(self, force: bool = False) -> None: """Poll hot/warm hrefs in the gaps between summary polls. Only runs in poll-only mode — in observe-primary mode those hrefs are - already covered by push notifications.""" - if self._observe.mode == MODE_OBSERVE: + already covered by push notifications — unless `force` is set, + which this cycle's sweep found disagreeing with the cache on a + still-live observe session (see log_sweep_discrepancies): a + bounded, self-limiting fallback for a channel that's gone silent + without a reconnect, without tearing down subscriptions that + would otherwise recover on their own once notifies resume.""" + if self._observe.mode == MODE_OBSERVE and not force: return hot = self._hot_hrefs warm = self._warm_hrefs if not hot and not warm: return - step = SUMMARY_INTERVAL_S / 10 # 3.0 s + step = self._SUBPOLL_STEP_S for i in range(1, 10): # slots 1..9 (T+3 s … T+27 s) await asyncio.sleep(step) hrefs = list(hot) + (list(warm) if i % 2 == 0 else []) @@ -208,7 +262,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._poll_hrefs_blocking, hrefs ) except Exception as e: - _LOGGER.debug("sub-poll batch failed: %s", e) + self._log.debug("sub-poll batch failed: %s", e) # ------------------------------------------------------------------ # Discovery (runs once on first successful poll) @@ -222,12 +276,12 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): reg = for_device(one_ui) if one_ui else None unbound: list[str] = [] if reg is not None: - _LOGGER.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui) + self._log.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui) bound = discover(resources, reg.capabilities, reg.pattern_capabilities, log=unbound.append) self.device_type_name = reg.name else: - _LOGGER.warning("unknown device type oneUiVersion=%r; using common caps", one_ui) + self._log.warning("unknown device type oneUiVersion=%r; using common caps", one_ui) bound = discover(resources, CAPABILITIES, log=unbound.append) self.device_type_name = None self.bound = bound @@ -265,7 +319,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._warm_hrefs = sorted(warm) self._discovered = True - _LOGGER.info( + self._log.info( "discovered %d entities (serial=%s) hot=%s warm=%s", len(bound), serial, self._hot_hrefs, self._warm_hrefs, ) @@ -307,7 +361,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): if sess is None: return await self.hass.async_add_executor_job( - self._observe.try_enter_observe_mode, sess, hrefs + self._observe.try_enter_observe_mode, sess, hrefs, + self._OBSERVE_GRACE_PERIOD_S, ) async def _maybe_retry_observe_mode(self) -> None: @@ -317,6 +372,34 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): return await self._attempt_observe_mode() + def _defer_reconnect_for(self, e: Exception) -> bool: + """True if this poll failure should NOT trigger a reconnect this + cycle. + + A `TimeoutError` (see `_poll_once`) means one block's ACK didn't + arrive in time — not that the session is dead. A recent OBSERVE + notify is direct proof the channel is still live, so always defer + in that case. Otherwise, defer until `_POLL_TIMEOUT_LIMIT` + consecutive timeouts have piled up — a single slow transfer is + normal on a flaky device; a run of them is a real problem. Any + other exception (a `ConnectionError`, an explicitly closed + session) is unambiguous and always reconnects immediately. + """ + if not isinstance(e, TimeoutError): + return False + if self._observe.mode == MODE_OBSERVE and self._observe.recently_notified(): + # Recent push is proof of life — reset the counter too, so + # timeouts from an earlier quiet stretch don't carry over and + # trigger a reconnect once the device goes quiet again. The + # counter should mean "consecutive timeouts with no push + # activity to vouch for the session," not just "consecutive + # timeouts" — otherwise an intermittently-active device could + # still accumulate its way into a false reconnect. + self._consecutive_poll_timeouts = 0 + return True + self._consecutive_poll_timeouts += 1 + return self._consecutive_poll_timeouts < self._POLL_TIMEOUT_LIMIT + # ------------------------------------------------------------------ # DataUpdateCoordinator hook # ------------------------------------------------------------------ @@ -327,22 +410,32 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._subpoll_task.cancel() self._subpoll_task = None + just_downgraded_from_observe = False async with self._session_lock: try: resources = await self.hass.async_add_executor_job(self._poll_once) + self._consecutive_poll_timeouts = 0 except Exception as e: + if self._defer_reconnect_for(e): + self._log.debug( + "poll failed (%s), not yet treated as session " + "death; skipping this cycle: %s", + type(e).__name__, e, + ) + return flatten(self.bound, self._cache.snapshot()) + self._consecutive_poll_timeouts = 0 # One reconnect attempt — pause briefly so the device can # clean up its DTLS session state before we knock again. - _LOGGER.warning("poll failed, reconnecting: %s", e) + self._log.warning("poll failed, reconnecting: %s", e) await self.hass.async_add_executor_job(self._close_session) - await asyncio.sleep(5.0) + await asyncio.sleep(self._RECONNECT_PAUSE_S) try: resources = await self.hass.async_add_executor_job(self._poll_once) except Exception as e2: - _LOGGER.error("poll failed after reconnect: %s", e2) + self._log.error("poll failed after reconnect: %s", e2) snapshot = self._cache.snapshot() if snapshot: - _LOGGER.debug("Full error:", exc_info=e2) + self._log.debug("Full error:", exc_info=e2) return flatten(self.bound, snapshot) raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 else: @@ -350,33 +443,50 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): # OBSERVE registrations. If we were in observe mode, # that state is now stale — the refresh task is still # pinned to the old (closed) session and nothing will - # ever re-subscribe on the new one. Tear it down now so - # the existing poll-mode retry path (_maybe_retry_observe_mode) - # re-subscribes on the live session on its next cycle. + # ever re-subscribe on the new one. Tear it down and + # try to resubscribe immediately below rather than + # waiting for the poll-mode retry timer — that timer + # exists to throttle devices that never had observe + # working at all, but a reconnect just proved this + # session is healthy, so there's no reason to wait. if self._observe.mode == MODE_OBSERVE: - _LOGGER.debug( + self._log.debug( "reconnect while in observe mode; downgrading to " - "poll so recovery re-subscribes on the new session" + "poll and resubscribing on the new session" ) self._observe.downgrade_to_poll() + just_downgraded_from_observe = True source = 'sweep' if self._discovered else 'poll' - if self._observe.mode == MODE_OBSERVE and self._observe.check_sweep_for_misses(resources): - self._observe.downgrade_to_poll() + sweep_mismatch = False + if self._observe.mode == MODE_OBSERVE: + # A sweep/cache mismatch never tears down a still-live OBSERVE + # session (see log_sweep_discrepancies) — the sweep below + # re-applies the authoritative state to the cache regardless, + # so there's nothing to correct by downgrading. Only a + # reconnect (above) proves subscriptions are actually gone. + # Instead, a mismatch triggers extra hot/warm subpolls this + # cycle below, so a channel gone silent without a reconnect + # (e.g. lost internet on an otherwise-live local session) + # still gets fresher-than-30s data. + sweep_mismatch = self._observe.log_sweep_discrepancies(resources) for href, rep in resources.items(): self._observe.apply(href, rep, source=source) if not self._discovered: self._run_discovery(resources) await self._attempt_observe_mode() + elif just_downgraded_from_observe: + await self._attempt_observe_mode() elif self._observe.mode == MODE_POLL: await self._maybe_retry_observe_mode() # Schedule sub-polls for hot/warm hrefs between summary polls - # (no-op in observe-primary mode; _run_subpolls checks the mode). + # (no-op in observe-primary mode unless this cycle's sweep found a + # mismatch; _run_subpolls checks the mode/force). if self._hot_hrefs or self._warm_hrefs: self._subpoll_task = self.hass.async_create_task( - self._run_subpolls(), name="localthings_subpoll" + self._run_subpolls(force=sweep_mismatch), name="localthings_subpoll" ) return flatten(self.bound, self._cache.snapshot()) @@ -396,7 +506,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): rep = self._cache.get(href or '') or {} result = write_fn(payload, rep, href) if result is None: - _LOGGER.warning("write_fn rejected payload %r for %s", payload, href) + self._log.warning("write_fn rejected payload %r for %s", payload, href) return path_segs, body = result @@ -406,11 +516,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): raise RuntimeError("no session") self._observe.mark_write_pending(href) code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0) - _LOGGER.info("PUT %s → code %#04x", href, code) + self._log.info("PUT %s → code %#04x", href, code) try: await self.hass.async_add_executor_job(_do_put) except Exception as e: - _LOGGER.error("command failed for %s: %s", href, e) + self._log.error("command failed for %s: %s", href, e) else: await self.async_request_refresh() diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index bf02edc..acda4fe 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -28,8 +28,32 @@ MODE_POLL = 'poll' DEFAULT_SETTLE_S = 4.0 GRACE_PERIOD_S = 15.0 + + +def _rep_diff(cached: dict, sweep: dict) -> dict: + """Fields present in BOTH reps whose values differ, as + {field: (cached_value, sweep_value)}. + + The batch `/device/0` sweep and an individual GET/notify can return + genuinely different-shaped reps for the same href (e.g. Samsung's + batch interface omits `rt`/`if` baseline fields that a direct GET + includes) — that's a representation-shape difference, not a missed + state change. Only comparing shared fields keeps the diff (and the + caller's miss detection) about real content changes. + """ + common = set(cached) & set(sweep) + return {k: (cached[k], sweep[k]) for k in common if cached[k] != sweep[k]} + + SUCCESS_FRACTION = 0.8 +# A notify within this long counts as proof the DTLS session is alive, +# even if the summary poll's blockwise GET just timed out on a slow +# device. Twice the coordinator's 30s summary interval: generous enough +# to absorb jitter, tight enough that a channel that's actually gone +# quiet doesn't get credit for a notify from a while ago. +PUSH_HEALTH_WINDOW_S = 60.0 + class ObserveManager: """Per-device observe-mode state: mode, write-settle guard, and (later) @@ -47,6 +71,7 @@ class ObserveManager: self._settle_lock = threading.Lock() self.subscribed_hrefs: set[str] = set() self._notified: set[str] = set() + self._last_notify_ts: float | None = None self.fallback_hrefs: set[str] = set() self._refresh_task: ObserveRefreshTask | None = None self._refresh_stop: threading.Event | None = None @@ -84,8 +109,25 @@ class ObserveManager: if not isinstance(rep, dict): return self._notified.add(href) + self._last_notify_ts = time.monotonic() + self.log.debug("observe notify: %s", href) self.apply(href, rep, source='observe') + def recently_notified(self, window_s: float = PUSH_HEALTH_WINDOW_S) -> bool: + """True if any OBSERVE notify has arrived within `window_s`. + + Used only to gate whether a poll (summary GET) failure should + close/reconnect the DTLS session — NOT to decide observe-mode + health in general (see `log_sweep_discrepancies` for why that + distinction matters: a notify recent enough to prove the session + itself is alive is a much stronger, narrower claim than "the + channel has been perfectly healthy"). + """ + return ( + self._last_notify_ts is not None + and time.monotonic() - self._last_notify_ts < window_s + ) + def try_enter_observe_mode( self, session, hrefs: list[str], grace_period_s: float = GRACE_PERIOD_S, @@ -130,25 +172,54 @@ class ObserveManager: self.last_mode_change_ts = time.monotonic() self.last_mode_change_wall = time.time() - def check_sweep_for_misses(self, sweep_resources: dict[str, dict]) -> bool: - """Compare a safety-net sweep result against the cache for every - subscribed href. A mismatch outside its settle window means - OBSERVE silently missed a change (it only notifies on change, so - any disagreement is a real miss, not a case of "nothing changed - yet"). Returns True if a miss was detected.""" + def log_sweep_discrepancies(self, sweep_resources: dict[str, dict]) -> bool: + """Log any subscribed href where the safety-net sweep disagrees + with the cache, on fields present in both reps (see `_rep_diff` + for why only shared fields count). Returns True if any + discrepancy was found (False if not in observe mode). + + This never changes mode or subscriptions. The sweep already + re-applies the full /device/0 result to the cache every cycle + regardless of mode, so a missed notify never leaves data stale + beyond one sweep interval — there's nothing to correct here. And + a still-live OBSERVE session should never be torn down over a + data-drift inference: some resources (e.g. an alarm's derived + "triggeredTime") appear to update without ever emitting a notify + even when the channel is otherwise perfectly healthy, so treating + a mismatch as proof of channel death produces false positives. + Downgrading also throws away working push coverage for every + *other* subscribed href just to react to one that isn't the + problem. + + The return value lets the caller respond more cheaply: without + an activity signal there's no way to tell "idle device, healthy + channel, nothing to notify about" from "dead channel, nothing + gets through" — both look the same here. So instead of guessing, + the coordinator uses a discrepancy as a trigger for extra hot/warm + subpolls this cycle (see `_run_subpolls`'s `force` parameter) — + a bounded, self-limiting response to a channel that's gone silent + for a reason OTHER than a reconnect (e.g. the device's OBSERVE + relay loses its internet connection while the local DTLS session + stays up), without tearing down subscriptions that will recover + on their own once notifies resume. + """ if self.mode != MODE_OBSERVE: return False + found = False for href in self.subscribed_hrefs: if href not in sweep_resources or self._is_settling(href): continue cached = self.cache.get(href) - if cached is not None and cached != sweep_resources[href]: - self.log.warning( - "observe missed a change on %s (sweep disagrees with cache)", - href, + if cached is None: + continue + diff = _rep_diff(cached, sweep_resources[href]) + if diff: + found = True + self.log.debug( + "observe missed a change on %s (sweep disagrees with cache): %s", + href, diff, ) - return True - return False + return found def downgrade_to_poll(self) -> None: self.fallback_hrefs = set(self.subscribed_hrefs) diff --git a/custom_components/localthings/registry/__init__.py b/custom_components/localthings/registry/__init__.py index c5bc00c..fe9adb6 100644 --- a/custom_components/localthings/registry/__init__.py +++ b/custom_components/localthings/registry/__init__.py @@ -6,8 +6,8 @@ to the matching capabilities, and flattens the result into HA-ready entity state. The DTLS/CoAP transport itself lives in the smartthings-local package, not here. """ -from .adapter import flatten, is_active, _key +from .adapter import flatten, _key from .discovery import BoundEntity, discover from .registry import CAPABILITIES -__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', 'is_active', '_key'] +__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', '_key'] diff --git a/custom_components/localthings/registry/adapter.py b/custom_components/localthings/registry/adapter.py index 92f379e..09bb574 100644 --- a/custom_components/localthings/registry/adapter.py +++ b/custom_components/localthings/registry/adapter.py @@ -22,13 +22,3 @@ def flatten(bound: list[BoundEntity], resources: dict) -> dict[str, Any]: elif b.desc.field: out[_key(b)] = b.desc.value_fn(rep.get(b.desc.field)) return out - - -def is_active(bound: list[BoundEntity], resources: dict) -> bool: - """Return True if any capability's active_when fires.""" - for b in bound: - if b.capability.active_when: - rep = resources.get(b.href) or {} - if b.capability.active_when(rep): - return True - return False diff --git a/custom_components/localthings/registry/capabilities/operational.py b/custom_components/localthings/registry/capabilities/operational.py index fb37ff3..c5ace01 100644 --- a/custom_components/localthings/registry/capabilities/operational.py +++ b/custom_components/localthings/registry/capabilities/operational.py @@ -28,10 +28,6 @@ def _int(v): return None -def _active_when(rep): - return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active' - - def _parse_hms(v): if not v: return None @@ -113,5 +109,4 @@ OPERATIONAL_STATE = Capability( ['operational', 'state', 'vs', '0'], {'x.com.samsung.da.state': p})), ), - active_when=_active_when, ) diff --git a/custom_components/localthings/registry/capabilities/oven.py b/custom_components/localthings/registry/capabilities/oven.py index cd36a69..4fc07af 100644 --- a/custom_components/localthings/registry/capabilities/oven.py +++ b/custom_components/localthings/registry/capabilities/oven.py @@ -213,11 +213,6 @@ def _naturalsteam_write(p, rep, href=None): return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts} -# --------------------------------------------------------------------------- -def _active_when(rep): - return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active' - - # --------------------------------------------------------------------------- # Capabilities # --------------------------------------------------------------------------- @@ -225,7 +220,6 @@ def _active_when(rep): OVEN_OPERATIONAL_STATE = Capability( href='/operational/state/vs/0', poll_tier='hot', - active_when=_active_when, entities=( SensorDesc(key='machine_state', field='x.com.samsung.da.state', name='Machine state', icon='mdi:stove', value_fn=_to_ocf), diff --git a/custom_components/localthings/registry/capability.py b/custom_components/localthings/registry/capability.py index b3d99a9..05921f3 100644 --- a/custom_components/localthings/registry/capability.py +++ b/custom_components/localthings/registry/capability.py @@ -16,7 +16,6 @@ class Capability: href_prefix: Optional[str] = None # pattern caps only: bind only if href starts with this strip_prefix_in_key: bool = False # strip href_prefix segs before building key_override match_fn: Optional[Callable[[dict, dict], bool]] = None # match_fn(rep, resources) -> bool - # Rare optional hooks — only operational-state-style resources use these. - active_when: Optional[Callable[[dict], bool]] = None + # Rare optional hook — only operational-state-style resources use this. on_observation: Optional[Callable[[dict, dict], None]] = None project: Optional[Callable[[dict, dict], dict]] = None diff --git a/tests/localthings/conftest.py b/tests/localthings/conftest.py index c19fc1e..4674f3c 100644 --- a/tests/localthings/conftest.py +++ b/tests/localthings/conftest.py @@ -35,6 +35,32 @@ from custom_components.localthings.const import ( CONF_CA_CERT_PEM, CONF_CA_KEY_PEM, CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN, ) +from custom_components.localthings.coordinator import LocalThingsCoordinator + + +@pytest.fixture(autouse=True) +def _fast_coordinator_timers(): + """Shrink the coordinator's real-time delays for every localthings test. + + `_run_subpolls` and `_attempt_observe_mode` are, by design, blocking/ + real-time operations in production (a real sub-poll cadence and a real + CoAP OBSERVE grace period). Tests that drive them through a full + `hass.config_entries.async_setup(...)` + `hass.async_block_till_done()` + would otherwise burn tens of real seconds per test waiting them out. + + This patches only the *class* attributes coordinator.py exposes for + this purpose (`_SUBPOLL_STEP_S`, `_OBSERVE_GRACE_PERIOD_S`) — the + production defaults they're computed from (`SUMMARY_INTERVAL_S`, + `GRACE_PERIOD_S`) are untouched. Tests that need to exercise the real + grace-period race (e.g. test_observe.py's own `ObserveManager` tests) + still pass `grace_period_s=` explicitly and are unaffected. + """ + with ( + patch.object(LocalThingsCoordinator, '_SUBPOLL_STEP_S', 0.001), + patch.object(LocalThingsCoordinator, '_OBSERVE_GRACE_PERIOD_S', 0.02), + patch.object(LocalThingsCoordinator, '_RECONNECT_PAUSE_S', 0.01), + ): + yield FIXTURES = Path(__file__).resolve().parent.parent / 'fixtures' diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index ee83309..34ffdcd 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -12,10 +12,10 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers import issue_registry as ir from custom_components.localthings.const import ( - DOMAIN, SUMMARY_INTERVAL_S, + CONF_HOST, DOMAIN, SUMMARY_INTERVAL_S, ) from custom_components.localthings.coordinator import LocalThingsCoordinator -from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL +from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL, PUSH_HEALTH_WINDOW_S from .conftest import ENTRY_DATA, MOCK_SERIAL @@ -182,6 +182,41 @@ async def test_last_resources_backed_by_state_cache( assert resources # fridge fixture is non-empty +def test_logger_is_scoped_to_device_host( + hass: HomeAssistant, mock_entry +) -> None: + """Every log line (coordinator's own, DataUpdateCoordinator's internal + messages, and ObserveManager's) must identify which device it's about — + a bare module-level logger is shared across every configured device and + makes multi-device logs ambiguous.""" + coordinator = LocalThingsCoordinator(hass, mock_entry) + + assert coordinator._log.name.endswith(ENTRY_DATA[CONF_HOST]) + assert coordinator.logger is coordinator._log + assert coordinator._observe.log is coordinator._log + + +def test_cache_changes_coalesce_into_a_single_push( + hass: HomeAssistant, mock_entry +) -> None: + """A poll/sweep cycle applying many hrefs in a tight loop must not + schedule one hass.add_job push per href — only one push per burst.""" + coordinator = LocalThingsCoordinator(hass, mock_entry) + + with patch.object(hass, 'add_job') as mock_add_job: + coordinator._on_cache_changed(True, 'poll') + coordinator._on_cache_changed(True, 'poll') + coordinator._on_cache_changed(True, 'poll') + + assert mock_add_job.call_count == 1 + + coordinator._push_cache_snapshot() + + coordinator._on_cache_changed(True, 'poll') + + assert mock_add_job.call_count == 2 + + async def test_enters_poll_mode_when_no_notifies_arrive( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -259,8 +294,21 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll( assert entered is True assert coordinator.observe_mode == MODE_OBSERVE + # Age out the notify so recently_notified() is False — a poll failure + # with a recent notify is now treated as evidence the session is + # still alive (see test_poll_failure_skips_reconnect_when_push_is_healthy) + # and would not trigger a reconnect at all. This test covers a + # genuinely dead channel: no recent push, poll fails, reconnect fires. + coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1 + # 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. with ( patch( 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', @@ -277,6 +325,332 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll( assert coordinator._observe.mode == MODE_POLL +async def test_poll_timeout_skips_reconnect_when_push_is_healthy( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """A poll (summary GET) TimeoutError with a recent OBSERVE notify must + not reconnect or downgrade at all — the notify proves the DTLS session + and its subscriptions are still alive, so the GET failure is just a + choked blockwise transfer on a slow device, not a dead channel. This + is the fix for a flaky device (e.g. a slow dishwasher) repeatedly + tearing down a healthy observe session on nothing but a poll hiccup.""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=TimeoutError('GET /device/0 block 11 timeout'), + ), + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._close_session', + ) as mock_close, + ): + await coordinator.async_request_refresh() + await hass.async_block_till_done() + + assert coordinator.observe_mode == MODE_OBSERVE + mock_close.assert_not_called() + + +async def test_poll_timeout_reconnects_after_consecutive_limit_even_with_push( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources +) -> None: + """Push health only matters at the moment of the failing poll — if + notifies later go stale too (not supplied here, so recently_notified() + becomes False once the window elapses) a run of consecutive timeouts + must still escalate to a reconnect, so a genuinely dead channel isn't + stuck forever just because it looked healthy once.""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + # No notify is recent any more, so every timeout below counts toward + # the consecutive-timeout limit instead of being deferred. + coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1 + + # Call _async_update_data directly rather than via + # async_request_refresh() — the coordinator's built-in debouncer + # (REQUEST_REFRESH_DEFAULT_COOLDOWN, real wall-clock seconds) would + # otherwise coalesce these rapid-fire calls into far fewer than the + # consecutive-timeout count this test needs to exercise. + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=TimeoutError('GET /device/0 block 11 timeout'), + ): + for _ in range(coordinator._POLL_TIMEOUT_LIMIT - 1): + await coordinator._async_update_data() + assert coordinator.observe_mode == MODE_OBSERVE + + # The final consecutive timeout crosses the limit and triggers the + # existing reconnect path, which then succeeds and downgrades. + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=[TimeoutError('GET /device/0 block 11 timeout'), fridge_resources], + ), + patch( + 'custom_components.localthings.coordinator.asyncio.sleep', + new=AsyncMock(), + ), + ): + await coordinator._async_update_data() + + assert coordinator.observe_mode == MODE_POLL + + +async def test_poll_timeout_counter_resets_when_push_is_healthy_again( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """Timeouts accrued during a quiet stretch must not survive into a + later burst of healthy push — otherwise an intermittently-active + device could reconnect on old timeouts even though push just proved + the session alive. The counter means "consecutive timeouts with no + push activity to vouch for the session," not just "consecutive + timeouts count.\"""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + # Age the notify out so timeouts accrue toward the limit. + coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1 + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=TimeoutError('GET /device/0 block 11 timeout'), + ): + for _ in range(coordinator._POLL_TIMEOUT_LIMIT - 1): + await coordinator._async_update_data() + assert coordinator._consecutive_poll_timeouts == coordinator._POLL_TIMEOUT_LIMIT - 1 + + # A fresh notify proves push is healthy again — the next timeout must + # not add to the pre-existing count, and must not reconnect. + fake.on_notification(hrefs[0], cbor2.dumps({'notified': True})) + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=TimeoutError('GET /device/0 block 11 timeout'), + ): + await coordinator._async_update_data() + + assert coordinator._consecutive_poll_timeouts == 0 + assert coordinator.observe_mode == MODE_OBSERVE + + +async def test_reconnect_from_observe_mode_resubscribes_immediately( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources +) -> None: + """A reconnect just proved the session is healthy, so the coordinator + must attempt resubscribe in the same update cycle rather than waiting + for the 600s poll-mode retry timer (_RECOVERY_RETRY_S) — that timer + exists to throttle devices that never had observe working at all, not + ones that just proved their session works.""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + # Age out the notify so recently_notified() is False and the poll + # failure below actually triggers a reconnect (see + # test_poll_failure_skips_reconnect_when_push_is_healthy for the + # healthy-push case, which now skips reconnecting entirely). + coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1 + + # If a stale retry timer (rather than an immediate resubscribe) were + # driving recovery, mode would still be 'poll' right after this single + # refresh call — the 600s gate wouldn't have elapsed yet. + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=[RuntimeError('connection lost'), fridge_resources], + ), + patch( + 'custom_components.localthings.coordinator.asyncio.sleep', + new=AsyncMock(), + ), + ): + notifier = _notify() + await coordinator.async_request_refresh() + await hass.async_block_till_done() + notifier.join() + + assert coordinator.observe_mode == MODE_OBSERVE + + +async def test_sweep_mismatch_never_downgrades_a_live_observe_session( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """A sweep/cache mismatch is diagnostic-only (see test_observe.py) — + it must never tear down a still-live OBSERVE session. The sweep + already re-applies the authoritative state to the cache regardless of + mode, so a downgrade would only throw away working push coverage on + every other subscribed href without fixing anything.""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + # A sweep result disagreeing with the cache on every subscribed href — + # previously treated as a device-wide miss and downgraded. + stale_sweep = {href: {'notified': False} for href in hrefs} + + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + return_value=stale_sweep, + ): + await coordinator.async_request_refresh() + await hass.async_block_till_done() + + assert coordinator.observe_mode == MODE_OBSERVE + # The mismatch is diagnostic-only, but the sweep result still won — + # the cache reflects the latest authoritative state either way. + assert coordinator.last_resources[hrefs[0]]['notified'] is False + + +async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """A sweep/cache mismatch on a live observe session must not tear + anything down, but it should trigger extra hot/warm subpolls this + cycle — the bounded fallback for a channel gone silent without a + reconnect (e.g. lost internet on an otherwise-live local session).""" + fake = mock_coordinator_observe_session + 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] + 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() + 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 + + stale_sweep = {href: {'notified': False} for href in hrefs} + + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + return_value=stale_sweep, + ), + patch.object( + LocalThingsCoordinator, '_run_subpolls', new_callable=AsyncMock, + ) as mock_subpolls, + ): + await coordinator.async_request_refresh() + await hass.async_block_till_done() + + mock_subpolls.assert_called_once_with(force=True) + + async def test_write_marks_href_pending_before_post( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py index c6cc789..a7e1aff 100644 --- a/tests/localthings/test_observe.py +++ b/tests/localthings/test_observe.py @@ -9,7 +9,9 @@ import pytest from smartthings_local.ocf.state_cache import StateCache -from custom_components.localthings.observe import ObserveManager, MODE_POLL +from custom_components.localthings.observe import ( + ObserveManager, MODE_POLL, _rep_diff, +) class _NullDescriptor: @@ -156,6 +158,25 @@ def test_on_notification_ignores_malformed_cbor(): assert mgr.cache.get('/oven/vs/0') is None +def test_recently_notified_false_before_any_notify(): + mgr = _manager() + assert mgr.recently_notified() is False + + +def test_recently_notified_true_just_after_a_notify(): + mgr = _manager() + mgr.on_notification('/oven/vs/0', cbor2.dumps({'a': 1})) + assert mgr.recently_notified() is True + + +def test_recently_notified_false_after_window_elapses(): + mgr = _manager() + mgr.on_notification('/oven/vs/0', cbor2.dumps({'a': 1})) + assert mgr.recently_notified(window_s=0.01) is True + time.sleep(0.02) + assert mgr.recently_notified(window_s=0.01) is False + + def test_try_enter_observe_mode_clears_stale_notifications_on_retry(): """Regression test: verify second call to try_enter_observe_mode() doesn't get polluted by notifications from the first call. This happens in real @@ -200,19 +221,65 @@ def test_try_enter_observe_mode_clears_stale_notifications_on_retry(): mgr.close() -def test_check_sweep_for_misses_false_when_not_in_observe_mode(): +def test_log_sweep_discrepancies_noop_when_not_in_observe_mode(caplog): mgr = _manager() - assert mgr.check_sweep_for_misses({'/oven/vs/0': {'a': 2}}) is False + with caplog.at_level('DEBUG'): + mgr.log_sweep_discrepancies({'/oven/vs/0': {'a': 2}}) + assert 'observe missed a change' not in caplog.text -def test_check_sweep_for_misses_detects_mismatch(): +def test_log_sweep_discrepancies_never_changes_mode(): + """A sweep/cache mismatch is diagnostic-only — a still-live OBSERVE + session is never torn down over a data-drift inference. The 30s sweep + already re-applies the authoritative state to the cache regardless of + mode, so there's nothing a downgrade would fix; it would only throw + away working push coverage on every other subscribed href.""" + mgr = _manager() + session = _FakeSession() + hrefs = ['/oven/vs/0', '/power/vs/0'] + + def _notify_during_grace_period(): + time.sleep(0.005) + for href in hrefs: + mgr.on_notification(href, cbor2.dumps({'a': 1})) + + notifier = threading.Thread(target=_notify_during_grace_period, daemon=True) + notifier.start() + + try: + mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02) + notifier.join() + assert mgr.mode == 'observe' + + # Sweep sees values the cache never got via notify on BOTH hrefs. + mgr.log_sweep_discrepancies({hrefs[0]: {'a': 2}, hrefs[1]: {'a': 2}}) + + assert mgr.mode == 'observe' + finally: + mgr.close() + + +def test_rep_diff_reports_only_shared_fields_that_differ(): + """Fields present on only one side (a shape difference, e.g. batch + sweep vs. individual GET/notify returning different fields for the + same href) must not be reported — only shared fields with different + values.""" + diff = _rep_diff({'a': 1, 'b': 2, 'c': 3}, {'a': 1, 'b': 5, 'd': 9}) + + assert diff == {'b': (2, 5)} + + +def test_log_sweep_discrepancies_ignores_shape_only_differences(caplog): + """A sweep rep missing/adding fields the cache doesn't have (or vice + versa) is a representation-shape difference, not a missed change — + must not be logged as a discrepancy.""" mgr = _manager() session = _FakeSession() href = '/oven/vs/0' def _notify_during_grace_period(): time.sleep(0.005) - mgr.on_notification(href, cbor2.dumps({'a': 1})) + mgr.on_notification(href, cbor2.dumps({'a': 1, 'rt': ['x'], 'if': ['y']})) notifier = threading.Thread(target=_notify_during_grace_period, daemon=True) notifier.start() @@ -222,15 +289,48 @@ def test_check_sweep_for_misses_detects_mismatch(): notifier.join() assert mgr.mode == 'observe' - # Sweep sees a value the cache never got via notify -> observe missed it. - missed = mgr.check_sweep_for_misses({href: {'a': 2}}) + # Sweep rep has the same 'a' value but lacks rt/if and adds 'href' — + # a shape difference (batch interface), not a missed content change. + with caplog.at_level('DEBUG'): + mgr.log_sweep_discrepancies({href: {'a': 1, 'href': href}}) + finally: + mgr.close() - assert missed is True + assert 'observe missed a change' not in caplog.text + + +def test_log_sweep_discrepancies_includes_the_diff(caplog): + """The log line must include what actually differed, not just that it + did.""" + mgr = _manager() + session = _FakeSession() + hrefs = ['/oven/vs/0', '/power/vs/0'] + + def _notify_during_grace_period(): + time.sleep(0.005) + for href in hrefs: + mgr.on_notification(href, cbor2.dumps({'a': 1, 'b': 2})) + + notifier = threading.Thread(target=_notify_during_grace_period, daemon=True) + notifier.start() + + try: + mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02) + notifier.join() + assert mgr.mode == 'observe' + + with caplog.at_level('DEBUG'): + mgr.log_sweep_discrepancies( + {hrefs[0]: {'a': 1, 'b': 5}, hrefs[1]: {'a': 1, 'b': 9}} + ) + + assert "{'b': (2, 5)}" in caplog.text + assert mgr.mode == 'observe' finally: mgr.close() -def test_check_sweep_for_misses_ignores_href_during_settle_window(): +def test_log_sweep_discrepancies_ignores_href_during_settle_window(caplog): mgr = _manager() session = _FakeSession() href = '/oven/vs/0' @@ -238,9 +338,10 @@ def test_check_sweep_for_misses_ignores_href_during_settle_window(): mgr.try_enter_observe_mode(session, [href], grace_period_s=0.01) mgr.mark_write_pending(href, settle_s=5.0) - missed = mgr.check_sweep_for_misses({href: {'a': 2}}) + with caplog.at_level('DEBUG'): + mgr.log_sweep_discrepancies({href: {'a': 2}}) - assert missed is False + assert 'observe missed a change' not in caplog.text def test_downgrade_to_poll_moves_subscribed_to_fallback(): diff --git a/tests/test_capability.py b/tests/test_capability.py index 9e704f1..25e738a 100644 --- a/tests/test_capability.py +++ b/tests/test_capability.py @@ -6,7 +6,6 @@ def test_capability_defaults(): c = Capability(href='/kidslock/vs/0', entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),)) assert c.poll_tier == 'cold' - assert c.active_when is None assert len(c.entities) == 1 diff --git a/tests/test_end_to_end_v2.py b/tests/test_end_to_end_v2.py index 36381bb..9a9ad80 100644 --- a/tests/test_end_to_end_v2.py +++ b/tests/test_end_to_end_v2.py @@ -2,7 +2,7 @@ import pytest from tests.conftest import _load_device from custom_components.localthings.registry.by_type import for_device, _type_key from custom_components.localthings.registry.discovery import discover -from custom_components.localthings.registry.adapter import flatten, is_active +from custom_components.localthings.registry.adapter import flatten @pytest.mark.parametrize('name,expected_type_key', [ @@ -24,5 +24,3 @@ def test_full_pipeline_v2(name, expected_type_key): state = flatten(bound, resources) assert state - # is_active returns bool; just verify it runs without error - is_active(bound, resources) diff --git a/tests/test_operational_capability.py b/tests/test_operational_capability.py index ec3adf8..70bc014 100644 --- a/tests/test_operational_capability.py +++ b/tests/test_operational_capability.py @@ -6,8 +6,3 @@ def test_machine_state_maps_samsung_to_ocf(): assert ms.value_fn('Run') == 'active' assert ms.value_fn('Pause') == 'pause' assert ms.value_fn('Ready') == 'idle' - - -def test_active_when_true_only_when_running(): - assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Run'}) is True - assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Ready'}) is False