test: address Copilot review on observe fixture typing and comment

- Parameterize FakeObserveSession.notify_on_subscribe as dict[str, Any] | None
  to match the surrounding fully-typed attributes (the dict is delivered as
  the OBSERVE rep to on_notification); add the typing.Any import.
- Reword the test_coordinator comment: subscribe() delivers a notification
  for every href it subscribes to (when notify_on_subscribe is set), not one.
This commit is contained in:
Tim Schindler
2026-07-22 08:22:45 +02:00
parent 0715eb0829
commit 8dadea6796
2 changed files with 15 additions and 10 deletions
+8 -5
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from unittest.mock import patch
import cbor2
@@ -156,11 +157,13 @@ class FakeObserveSession:
# 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 lands inside the grace period: a notify raced in from
# another thread can be wiped by that clear (or arrive after the
# grace period ends) depending on scheduling. Set to None to model
# a device that answers subscriptions but never notifies.
self.notify_on_subscribe: dict | None = None
# 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)
+7 -5
View File
@@ -258,11 +258,13 @@ 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. The
# fake session delivers one synchronously from subscribe() (see
# FakeObserveSession.notify_on_subscribe in conftest.py), which puts
# them inside that window by construction rather than by timing.
# 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,