diff --git a/.github/release-notes/2.0.0b16.md b/.github/release-notes/2.0.0b16.md index a871f4e95..737ab833d 100644 --- a/.github/release-notes/2.0.0b16.md +++ b/.github/release-notes/2.0.0b16.md @@ -20,6 +20,7 @@ See the cover docs for the details and the known limits. - **`myhome.send_message` and `myhome.sync_time` reject unknown keys.** A misspelt key such as `gw_mac` or a missing `message` raises a validation error instead of sending the frame to the default gateway; both services get text selectors in the visual editor (#189). - **Climate zones show the setpoint in force (dimension 12).** When the thermostat regulates to another setpoint than the commanded one (dimension 14; 23.5 in force while 21.0 commanded on an F461), the entity and `local_target_temperature` now show the one in force instead of the commanded value plus the knob offset. Known limit: once a zone has reported dimension 12, a dimension 14 that is not the echo of a Home Assistant write does not move the slider until the next dimension 12 report or poll (see Known Limitations). +- **Event session liveness probe.** After 10 minutes without a bus frame on the event session, `*#13**0##` goes out on the command session; gateways that echo the reply on the event session (F461) have the session recreated when the echo is missing, and a repair issue `event_session_silent` is raised after three misses in a row. Event-triggered zone polls no longer block the event reader on a full send queue (#123, #135, #158, #164). - **`myhome.sync_time` refuses time zones with a half-hour offset** (e.g. `Asia/Kolkata`, UTC+5:30) with a translated error. The WHO 13 time-zone field carries whole hours only; b15 (on OWNd 2.0.0b10) set the gateway clock with a truncated offset without saying so (#764). - **`advanced: true` on a standard F411 shutter now logs a startup warning.** Those actuators have no position reporting, so the position stayed at 50 % and calibration was refused; the error now says to remove `advanced: true` (#761, #762). diff --git a/custom_components/myhome/gateway.py b/custom_components/myhome/gateway.py index 8c267560a..3abd1cb8f 100644 --- a/custom_components/myhome/gateway.py +++ b/custom_components/myhome/gateway.py @@ -655,6 +655,16 @@ def claim_in_flight_dimension_where(self, who: int, dimension: int) -> str | Non return standby._command_pool.claim_in_flight_dimension_where(who, dimension) return self._command_pool.claim_in_flight_dimension_where(who, dimension) + def send_status_request_nowait(self, message: OWNCommand) -> asyncio.Future[float] | None: + """Queue a status request without blocking; None when the send queue is full.""" + delegated = self._delegated_target(message) + if delegated is not None: + return delegated.send_status_request_nowait(message) + standby = self._failover_target(message) + if standby is not None: + return standby.send_status_request_nowait(message) + return self._command_pool.send_status_request_nowait(message) + def _known_light_areas(self) -> list[str]: """Return list of known light areas.""" return self._resync_manager.known_light_areas() diff --git a/custom_components/myhome/gateway_events.py b/custom_components/myhome/gateway_events.py index 10916c05d..aad08fcc5 100644 --- a/custom_components/myhome/gateway_events.py +++ b/custom_components/myhome/gateway_events.py @@ -407,7 +407,13 @@ async def process_message(self, message: Any) -> None: ) from OWNd.message import OWNHeatingCommand as OWNHCmd - await self.handler.send_status_request(OWNHCmd.status(where)) + # Never block the event read loop on a full send queue: skip the poll instead. + if self.handler.send_status_request_nowait(OWNHCmd.status(where)) is None: + self._logger.debug( + "%s Send queue full; zone %s status poll skipped", + self.handler.log_id, + where, + ) elif isinstance(message, OWNCENPlusEvent): event = None if message.is_short_pressed: diff --git a/custom_components/myhome/gateway_sessions.py b/custom_components/myhome/gateway_sessions.py index 62439b9f4..fbcb6da99 100644 --- a/custom_components/myhome/gateway_sessions.py +++ b/custom_components/myhome/gateway_sessions.py @@ -18,7 +18,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_send from OWNd.connection import OWNCommandSession, OWNEventSession, OWNGateway -from OWNd.message import OWNCommand, OWNMessage +from OWNd.message import OWNCommand, OWNGatewayCommand, OWNGatewayEvent, OWNMessage, OWNSignaling from .const import LOGGER, area_of_where, is_apl_address from .expose import is_exposed_on_bus @@ -40,6 +40,31 @@ # leaving sessions open indefinitely. Installations desiring a generous hard ceiling # for "TCP up, bus dead" conditions can override this constant (e.g. 43200.0 / 12h). EVENT_INACTIVITY_TIMEOUT: float | None = None +# End-to-end liveness probe (#123, #135, #158, #164): after EVENT_PROBE_IDLE seconds +# without a bus frame (keepalive ACKs do not count), `*#13**0##` goes out on the +# command session and its reply must also arrive on the event session within +# EVENT_PROBE_TIMEOUT (live F461 2.0.11). A miss recycles the event session; after +# EVENT_PROBE_REPAIR_AFTER consecutive misses a repair issue is raised. A gateway is +# only recycled once it has echoed a probe, or when its model is known to echo, +# so a gateway that keeps the reply on the command session is never cycled. +EVENT_PROBE_IDLE: float = 600.0 +EVENT_PROBE_TIMEOUT: float = 30.0 +EVENT_PROBE_REPAIR_AFTER: int = 3 +# Models seen to deliver the ``*#13**0##`` reply on the event session (physical F461 +# 2.0.11). Fallback for OWNd profiles without an ``echoes_event_session_probe`` flag; +# the flag belongs in OWNd (#381 pattern). +EVENT_PROBE_ECHO_MODELS = frozenset({"F461"}) + + +def _is_time_echo(message: object) -> bool: + """Whether ``message`` is the gateway's ``*#13**0*…##`` time reply, the probe's echo. + + OWNd parses it as an ``OWNGatewayEvent`` with dimension 0; the raw-string prefix is + the fallback for a reply the pinned OWNd hands back unparsed. + """ + if isinstance(message, OWNGatewayEvent): + return message.dimension == 0 + return str(message).startswith("*#13**0*") def _resolve_written(task: dict[str, Any], when: float) -> None: @@ -82,6 +107,13 @@ def __init__( self._event_session_ready: asyncio.Event = asyncio.Event() self._event_watchdog: asyncio.Timeout | None = None self.is_connected: bool = False + self._last_rx_at: float = time.monotonic() + self._last_echo_at: float = 0.0 + self._last_probe_at: float = 0.0 + self._rx_event: asyncio.Event = asyncio.Event() + self._probe_echo_seen: bool = False + self._probe_misses: int = 0 + self._probe_recycle: bool = False @property def gateway(self) -> OWNGateway: @@ -110,7 +142,8 @@ def _update_event_watchdog(self, *, progress: bool) -> None: a bare state change only arms it when it is not already running. """ watchdog = self._event_watchdog - if watchdog is None or watchdog.expired(): + if watchdog is None or watchdog.expired() or self._probe_recycle: + # A recycle requested by the liveness probe must not be disarmed by a late frame. return if self.is_connected: watchdog.reschedule(None) @@ -192,7 +225,9 @@ async def _run_event_session(self) -> bool: await self._read_event_session(_event_session) return False except Exception as err: - if isinstance(err, TimeoutError) and watchdog.expired(): + if isinstance(err, TimeoutError) and watchdog.expired() and self._probe_recycle: + LOGGER.debug("%s Event session closed by the liveness probe.", self.log_id) + elif isinstance(err, TimeoutError) and watchdog.expired(): LOGGER.warning( "%s Event session stalled: disconnected with no reconnect " "progress for %ss.", @@ -203,6 +238,7 @@ async def _run_event_session(self) -> bool: LOGGER.exception("%s Event listener failed.", self.log_id) finally: self._event_watchdog = None + self._probe_recycle = False with contextlib.suppress(Exception): await asyncio.shield(_event_session.close()) return not self._terminate_listener @@ -245,11 +281,28 @@ async def _read_event_session(self, _event_session: OWNEventSession) -> None: self.log_id, ) self._update_event_watchdog(progress=True) + self._last_rx_at = time.monotonic() + probe = asyncio.get_running_loop().create_task( + self._liveness_probe(), name=f"myhome_{self.handler.mac}_event_probe" + ) + try: + await self._dispatch_event_frames(_event_session, session_is_open) + finally: + probe.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await probe + async def _dispatch_event_frames(self, _event_session: OWNEventSession, session_is_open: Any) -> None: + """Read ``_event_session`` until the listener terminates or the read raises.""" was_reachable = True while not self._terminate_listener: message = await _event_session.get_next() self._update_event_watchdog(progress=True) + if message is not None and not isinstance(message, OWNSignaling): + self._last_rx_at = time.monotonic() + if _is_time_echo(message): + self._last_echo_at = self._last_rx_at + self._rx_event.set() if message is None: reachable = session_is_open(_event_session) if reachable != was_reachable: @@ -277,6 +330,106 @@ async def _read_event_session(self, _event_session: OWNEventSession) -> None: except Exception: LOGGER.exception("%s Failed to process `%s`.", self.log_id, message) + async def _liveness_probe(self) -> None: + """Probe a silent event session end to end, for as long as it is open.""" + while not self._terminate_listener: + idle = float(EVENT_PROBE_IDLE) + delay = max(self._last_rx_at, self._last_probe_at) + idle - time.monotonic() + if delay > 0: + await asyncio.sleep(delay) + continue + if not self.is_connected: + # A lost socket is OWNd's and the stall watchdog's business. + self._last_probe_at = time.monotonic() + continue + try: + if not await self._probe_once(): + return + except Exception: + LOGGER.exception("%s Event session liveness probe failed.", self.log_id) + + def _probe_echo_expected(self) -> bool: + """Whether a missing echo means a dead event session on this gateway. + + A seen echo is the strongest evidence; then the OWNd profile flag + ``echoes_event_session_probe`` (``None`` = unverified); then the model table. + """ + profile = getattr(self.handler, "profile", None) + flag = getattr(profile, "echoes_event_session_probe", None) + model = str(getattr(self.handler, "model", "") or "").strip().upper() + if self._probe_echo_seen: + return True + if isinstance(flag, bool): + return flag + return model in EVENT_PROBE_ECHO_MODELS + + async def _probe_once(self) -> bool: + """Send one probe; False when it was not echoed and the session is being recycled.""" + self._last_probe_at = time.monotonic() + silent_for = self._last_probe_at - self._last_rx_at + # Never wait for room in a full queue: the probe is retried after the next idle period. + written = self.handler._command_pool.send_status_request_nowait(OWNGatewayCommand.request_time()) + if written is not None: + await asyncio.wait({written}, timeout=float(EVENT_PROBE_TIMEOUT)) + if written is None or not written.done() or written.cancelled() or written.exception() is not None: + LOGGER.debug("%s Liveness probe not sent: command queue full or session unavailable.", self.log_id) + return True + written_at = written.result() + deadline = written_at + float(EVENT_PROBE_TIMEOUT) + while self._last_rx_at <= written_at and (remaining := deadline - time.monotonic()) > 0: + self._rx_event.clear() + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(self._rx_event.wait(), timeout=remaining) + if self._last_rx_at > written_at: + # Any frame proves the event path; only the time reply proves this gateway echoes. + if self._last_echo_at > written_at: + self._probe_echo_seen = True + if self._probe_misses: + LOGGER.info("%s Event session answers the liveness probe again.", self.log_id) + self._delete_probe_issue() + self._probe_misses = 0 + return True + if not self._probe_echo_expected(): + LOGGER.debug( + "%s `*#13**0##` was not echoed on the event session; this gateway has never " + "echoed it, so the session is kept.", + self.log_id, + ) + return True + self._probe_misses += 1 + LOGGER.warning( + "%s Event session silent for %.0fs and `*#13**0##` was not echoed within %.0fs " + "(miss %d); recreating the event session.", + self.log_id, + silent_for, + float(EVENT_PROBE_TIMEOUT), + self._probe_misses, + ) + if self._probe_misses >= int(EVENT_PROBE_REPAIR_AFTER): + self._create_probe_issue() + self._probe_recycle = True + watchdog = self._event_watchdog + if watchdog is not None and not watchdog.expired(): + watchdog.reschedule(asyncio.get_running_loop().time()) + return False + + def _create_probe_issue(self) -> None: + from .repairs import async_create_event_session_silent_issue + + entry = getattr(self.handler, "config_entry", None) + entry_id = getattr(entry, "entry_id", None) + if isinstance(entry_id, str): + async_create_event_session_silent_issue( + self.handler.hass, entry_id, str(getattr(entry, "title", "") or self.handler.mac), self._probe_misses + ) + + def _delete_probe_issue(self) -> None: + from .repairs import async_delete_event_session_silent_issue + + entry_id = getattr(getattr(self.handler, "config_entry", None), "entry_id", None) + if isinstance(entry_id, str): + async_delete_event_session_silent_issue(self.handler.hass, entry_id) + def close(self) -> None: """Signal listener termination and unblock waiting workers.""" self._terminate_listener = True @@ -1217,15 +1370,28 @@ async def send_status_request(self, message: OWNCommand) -> asyncio.Future[float """Queue a status request; the returned future resolves to the monotonic write time.""" return await self._enqueue(message, is_status_request=True) - async def _enqueue( - self, + def send_status_request_nowait(self, message: OWNCommand) -> asyncio.Future[float] | None: + """Queue a status request without waiting for room; None when the queue is full or closed. + + For callers on the event read loop, which must never block on a full queue. + """ + envelope = self._envelope(message, is_status_request=True) + try: + self.send_buffer.put_nowait(envelope) + except (asyncio.QueueFull, getattr(asyncio, "QueueShutDown", asyncio.QueueFull)): + envelope["written"].cancel() + return None + return envelope["written"] # type: ignore[no-any-return] + + @staticmethod + def _envelope( message: OWNCommand, *, is_status_request: bool, is_fade_step: bool = False, is_cancel: bool = False, - ) -> asyncio.Future[float]: - """Put a frame on the send queue and hand back its delivery future.""" + ) -> dict[str, Any]: + """The queue item for ``message``, with a fresh delivery future.""" written: asyncio.Future[float] = asyncio.get_running_loop().create_future() envelope: dict[str, Any] = { "message": message, @@ -1235,6 +1401,21 @@ async def _enqueue( } if is_fade_step: envelope["is_fade_step"] = True + return envelope + + async def _enqueue( + self, + message: OWNCommand, + *, + is_status_request: bool, + is_fade_step: bool = False, + is_cancel: bool = False, + ) -> asyncio.Future[float]: + """Put a frame on the send queue and hand back its delivery future.""" + envelope = self._envelope( + message, is_status_request=is_status_request, is_fade_step=is_fade_step, is_cancel=is_cancel + ) + written: asyncio.Future[float] = envelope["written"] await self.send_buffer.put(envelope) LOGGER.debug( "%s Message `%s` was successfully queued.", diff --git a/custom_components/myhome/repairs.py b/custom_components/myhome/repairs.py index 480c1d7f1..553a7804b 100644 --- a/custom_components/myhome/repairs.py +++ b/custom_components/myhome/repairs.py @@ -42,6 +42,7 @@ ISSUE_GATEWAY_IDENTITY = "gateway_identity_mismatch" ISSUE_UNKNOWN_GATEWAY_MODEL = "unknown_gateway_model" ISSUE_UNCONFIGURED_TIMEZONE = "unconfigured_timezone" +ISSUE_EVENT_SESSION_SILENT = "event_session_silent" ISSUE_GATEWAY_IDENTITY_CORRECTED = "gateway_identity_corrected" ISSUE_INCOMPATIBLE_DECODER = "incompatible_decoder_platform" @@ -104,6 +105,26 @@ def async_delete_unconfigured_timezone_issue(hass: HomeAssistant, entry_id: str) async_delete_issue(hass, DOMAIN, f"{ISSUE_UNCONFIGURED_TIMEZONE}_{entry_id}") +def async_create_event_session_silent_issue( + hass: HomeAssistant, entry_id: str, gateway_name: str, misses: int +) -> None: + """Raise a repair when the event session repeatedly fails the end-to-end liveness probe.""" + async_create_issue( + hass, + DOMAIN, + f"{ISSUE_EVENT_SESSION_SILENT}_{entry_id}", + is_fixable=False, + severity=IssueSeverity.WARNING, + translation_key=ISSUE_EVENT_SESSION_SILENT, + translation_placeholders={"gateway": gateway_name, "misses": str(misses)}, + ) + + +def async_delete_event_session_silent_issue(hass: HomeAssistant, entry_id: str) -> None: + """Delete the silent event session issue once a probe is echoed again.""" + async_delete_issue(hass, DOMAIN, f"{ISSUE_EVENT_SESSION_SILENT}_{entry_id}") + + def async_create_identity_issue( hass: HomeAssistant, entry_id: str, configured: str, reported: str, code: str, source: str, official: bool ) -> None: diff --git a/custom_components/myhome/strings.json b/custom_components/myhome/strings.json index 13c79d4e3..d96d02ffc 100644 --- a/custom_components/myhome/strings.json +++ b/custom_components/myhome/strings.json @@ -453,6 +453,10 @@ "title": "Authentication failed for MyHOME gateway {gateway}", "description": "The MyHOME gateway refused the configured OpenWebNet password or HMAC negotiation. Please check your credentials using the reconfigure flow." }, + "event_session_silent": { + "title": "MyHOME gateway {gateway} stopped reporting bus events", + "description": "The event connection to {gateway} stayed open, but a gateway time request sent on the command connection was not echoed on it {misses} times in a row, even after the event connection was recreated. Home Assistant may miss light, cover and sensor changes.\n\nCheck that no other client exhausts the gateway's sessions and restart the gateway. This issue clears itself when the gateway answers again." + }, "unconfigured_timezone": { "title": "Unconfigured timezone on MyHOME gateway ({gateway})", "description": "The MyHOME gateway {gateway} is reporting an unconfigured timezone (placeholder code '999'). This can cause date and time parsing failures or dropped gateway diagnostic messages.\n\nPlease log into the gateway web UI or MyHOME Suite and configure a valid timezone, then restart the gateway." diff --git a/custom_components/myhome/translations/en.json b/custom_components/myhome/translations/en.json index 13c79d4e3..d96d02ffc 100644 --- a/custom_components/myhome/translations/en.json +++ b/custom_components/myhome/translations/en.json @@ -453,6 +453,10 @@ "title": "Authentication failed for MyHOME gateway {gateway}", "description": "The MyHOME gateway refused the configured OpenWebNet password or HMAC negotiation. Please check your credentials using the reconfigure flow." }, + "event_session_silent": { + "title": "MyHOME gateway {gateway} stopped reporting bus events", + "description": "The event connection to {gateway} stayed open, but a gateway time request sent on the command connection was not echoed on it {misses} times in a row, even after the event connection was recreated. Home Assistant may miss light, cover and sensor changes.\n\nCheck that no other client exhausts the gateway's sessions and restart the gateway. This issue clears itself when the gateway answers again." + }, "unconfigured_timezone": { "title": "Unconfigured timezone on MyHOME gateway ({gateway})", "description": "The MyHOME gateway {gateway} is reporting an unconfigured timezone (placeholder code '999'). This can cause date and time parsing failures or dropped gateway diagnostic messages.\n\nPlease log into the gateway web UI or MyHOME Suite and configure a valid timezone, then restart the gateway." diff --git a/docs/diagnostics/repair-issues.md b/docs/diagnostics/repair-issues.md index b3309a3b0..a89c7c8d9 100644 --- a/docs/diagnostics/repair-issues.md +++ b/docs/diagnostics/repair-issues.md @@ -41,6 +41,24 @@ Once the gateway broadcasts an updated WHO=13 frame with a valid timezone offset --- +## Event Session Silent + +**Repair Key**: `event_session_silent` +**Severity**: `WARNING` +**Auto-Resolving**: Yes + +### What it means +When no bus frame arrived on the event session for 10 minutes, MyHOME sends the gateway time request `*#13**0##` on the command session. Gateways such as the F461 also deliver the reply on the event session, which proves the event path end to end. When that echo is missing, the event session is recreated; when it is still missing after three attempts in a row, this issue is raised. A gateway is only recycled once it has echoed the probe at least once (or is an F461), so gateways that keep the reply on the command session are never affected. + +### How to resolve +1. Check that no other client (another Home Assistant, a supervisor app, MyHOME_Suite) holds all of the gateway's sessions. +2. Restart the gateway. + +### How it clears +The issue is withdrawn as soon as a probe is echoed again. + +--- + ## Unknown Gateway Model **Repair Key**: `unknown_gateway_model` diff --git a/tests/test_event_session_liveness.py b/tests/test_event_session_liveness.py new file mode 100644 index 000000000..4ff8244b5 --- /dev/null +++ b/tests/test_event_session_liveness.py @@ -0,0 +1,407 @@ +"""End-to-end event-session liveness probe and a non-blocking event read loop (#123, #135, #158, #164).""" +import asyncio +import logging +import time +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from OWNd.message import OWNMessage + +from custom_components.myhome import gateway_sessions +from custom_components.myhome.gateway import MyHOMEGatewayHandler +from custom_components.myhome.gateway_sessions import FrameScheduler + +PROBE_ECHO = "*#13**0*20*42*07*001##" + + +def _handler(model: str) -> MyHOMEGatewayHandler: + entry = MagicMock() + entry.entry_id = "entry_probe" + entry.title = f"{model} Gateway" + entry.data = { + "host": "192.0.2.92", + "port": 20000, + "password": "12345", + "mac": "00:03:50:00:07:77", + "name": model, + "manufacturer": "BTicino", + "firmware": "2.0.11", + } + entry.options = {} + hass = MagicMock() + hass.data = {} + return MyHOMEGatewayHandler(hass, entry) + + +@pytest.fixture +def fast_probe(monkeypatch): + monkeypatch.setattr(gateway_sessions, "EVENT_PROBE_IDLE", 0.05) + monkeypatch.setattr(gateway_sessions, "EVENT_PROBE_TIMEOUT", 0.1) + monkeypatch.setattr(gateway_sessions, "EVENT_PROBE_REPAIR_AFTER", 2) + monkeypatch.setattr("custom_components.myhome.gateway.EVENT_RESTART_BACKOFF_MIN", 0.01) + monkeypatch.setattr("custom_components.myhome.gateway.EVENT_RESTART_BACKOFF_MAX", 0.02) + + +def _silent_session(frames: asyncio.Queue | None = None) -> MagicMock: + """A session that connects and then never delivers a frame unless one is queued.""" + session = MagicMock() + session.connect = AsyncMock(return_value={"Success": True}) + session.close = AsyncMock() + queue = frames if frames is not None else asyncio.Queue() + + async def get_next(): + return await queue.get() + + session.get_next = AsyncMock(side_effect=get_next) + return session + + +def _written_now(*_args, **_kwargs): + fut = asyncio.get_running_loop().create_future() + fut.set_result(time.monotonic()) + return fut + + +async def _run_listener_until(handler, condition, timeout=3.0): + task = asyncio.create_task(handler.listening_loop()) + try: + async with asyncio.timeout(timeout): + while not condition(): + await asyncio.sleep(0.01) + finally: + handler._event_runner.close() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_connected_but_silent_session_is_recycled_and_repaired(fast_probe, caplog): + """No echo of `*#13**0##` on a gateway known to echo it: the session is recreated, then a repair.""" + handler = _handler("F461") + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session()) + return sessions[-1] + + probe_send = MagicMock(side_effect=_written_now) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + patch("custom_components.myhome.repairs.async_create_event_session_silent_issue") as create_issue, + caplog.at_level(logging.WARNING, logger="custom_components.myhome"), + ): + await _run_listener_until(handler, lambda: create_issue.called and len(sessions) >= 3) + + assert str(probe_send.call_args_list[0].args[0]) == "*#13**0##" + assert len(sessions) >= 3 + for session in sessions[:2]: + session.close.assert_awaited() + assert any("was not echoed" in r.getMessage() for r in caplog.records) + assert not any("stalled" in r.getMessage() for r in caplog.records) + create_issue.assert_called_with(handler.hass, "entry_probe", "F461 Gateway", 2) + + +async def test_echoed_probe_keeps_the_session(fast_probe): + """The echo arrives on the event session: no recycle, and the gateway is marked as echoing.""" + handler = _handler("MH202") + frames: asyncio.Queue = asyncio.Queue() + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session(frames)) + return sessions[-1] + + def send_and_echo(message): + frames.put_nowait(OWNMessage.parse(PROBE_ECHO)) + return _written_now() + + probe_send = MagicMock(side_effect=send_and_echo) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + ): + await _run_listener_until(handler, lambda: probe_send.call_count >= 3) + + assert len(sessions) == 1 + assert handler._event_runner._probe_echo_seen is True + assert handler._event_runner._probe_misses == 0 + + +async def test_unrelated_traffic_keeps_the_session_but_does_not_prove_an_echo(fast_probe): + """A light frame during the probe window proves the event path, not that the gateway echoes `*#13**0##`.""" + handler = _handler("MH202") + frames: asyncio.Queue = asyncio.Queue() + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session(frames)) + return sessions[-1] + + def send_and_light(message): + frames.put_nowait(OWNMessage.parse("*1*1*06##")) + return _written_now() + + probe_send = MagicMock(side_effect=send_and_light) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + ): + await _run_listener_until(handler, lambda: probe_send.call_count >= 2) + + assert len(sessions) == 1 + assert handler._event_runner._probe_echo_seen is False + + +async def test_gateway_that_never_echoed_is_not_recycled(fast_probe): + """Without evidence that this gateway echoes the probe, a missing echo proves nothing.""" + handler = _handler("MH202") + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session()) + return sessions[-1] + + probe_send = MagicMock(side_effect=_written_now) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + ): + await _run_listener_until(handler, lambda: probe_send.call_count >= 3) + + assert len(sessions) == 1 + + +async def test_keepalive_acks_do_not_hide_a_dead_bus(fast_probe): + """`*#*1##` ACKs to the profile keepalive keep TCP alive but are not bus traffic.""" + handler = _handler("F461") + frames: asyncio.Queue = asyncio.Queue() + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session(frames)) + return sessions[-1] + + async def ack_pump(): + while True: + frames.put_nowait(OWNMessage.parse("*#*1##")) + await asyncio.sleep(0.01) + + pump = asyncio.create_task(ack_pump()) + try: + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", MagicMock(side_effect=_written_now)), + patch("custom_components.myhome.repairs.async_create_event_session_silent_issue"), + ): + await _run_listener_until(handler, lambda: len(sessions) >= 2) + finally: + pump.cancel() + + assert len(sessions) >= 2 + + +async def test_unsent_probe_is_not_a_miss(fast_probe): + """A probe the command session could not write says nothing about the event session.""" + handler = _handler("F461") + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session()) + return sessions[-1] + + def not_written(*_args): + fut = asyncio.get_running_loop().create_future() + fut.cancel() + return fut + + probe_send = MagicMock(side_effect=not_written) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + ): + await _run_listener_until(handler, lambda: probe_send.call_count >= 3) + + assert len(sessions) == 1 + assert handler._event_runner._probe_misses == 0 + + +async def test_full_send_queue_does_not_block_the_event_reader(caplog): + """A setpoint echo (`*#4*1*#14*0210*3##`) polls the zone without waiting for room in a full queue.""" + handler = _handler("F461") + handler.send_buffer = FrameScheduler(maxsize=1) + handler.send_buffer.put_nowait({"message": "*1*1*06##", "is_status_request": False, "written": None}) + + with caplog.at_level(logging.DEBUG, logger="custom_components.myhome"): + async with asyncio.timeout(1): + await handler._process_message(OWNMessage.parse("*#4*1*#14*0210*3##")) + + assert handler.send_buffer.qsize() == 1 + assert any("status poll skipped" in r.getMessage() for r in caplog.records) + + +async def test_event_triggered_poll_is_queued_when_there_is_room(): + handler = _handler("F461") + await handler._process_message(OWNMessage.parse("*#4*1*#14*0210*3##")) + + queued = handler.send_buffer.get_nowait() + assert str(queued["message"]) == "*#4*1##" + assert queued["is_status_request"] is True + + +@pytest.mark.parametrize( + ("model", "flag", "expected"), + [("MH202", True, True), ("F461", False, False), ("F461", None, True), ("MH202", None, False)], +) +def test_profile_flag_decides_before_the_model_table(model, flag, expected): + """`echoes_event_session_probe` on the OWNd profile wins; `None` (unverified) falls back to the table.""" + handler = _handler(model) + profile = MagicMock() + profile.echoes_event_session_probe = flag + with patch.object(type(handler), "profile", new_callable=lambda: property(lambda self: profile)): + assert handler._event_runner._probe_echo_expected() is expected + + +async def test_probe_waits_while_the_session_is_disconnected(monkeypatch): + """A lost socket is OWNd's and the stall watchdog's business: no probe goes out.""" + monkeypatch.setattr(gateway_sessions, "EVENT_PROBE_IDLE", 0.01) + handler = _handler("F461") + runner = handler._event_runner + runner.is_connected = False + runner._last_rx_at = time.monotonic() - 1 + probe_once = AsyncMock() + with patch.object(runner, "_probe_once", probe_once): + task = asyncio.create_task(runner._liveness_probe()) + await asyncio.sleep(0.05) + runner.close() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + probe_once.assert_not_awaited() + assert runner._last_probe_at > runner._last_rx_at + + +async def test_probe_failure_is_logged_and_the_probe_keeps_running(monkeypatch, caplog): + """An exception inside one probe does not end the probe for the life of the session.""" + monkeypatch.setattr(gateway_sessions, "EVENT_PROBE_IDLE", 0.01) + handler = _handler("F461") + runner = handler._event_runner + runner.is_connected = True + calls = 0 + + async def flaky(): + nonlocal calls + calls += 1 + runner._last_probe_at = time.monotonic() + if calls == 1: + raise RuntimeError("boom") + runner._terminate_listener = True + return True + + with patch.object(runner, "_probe_once", flaky), caplog.at_level(logging.ERROR, logger="custom_components.myhome"): + async with asyncio.timeout(2): + await runner._liveness_probe() + + assert calls == 2 + assert any("liveness probe failed" in r.getMessage() for r in caplog.records) + + +async def test_echo_after_a_miss_withdraws_the_repair(fast_probe, caplog): + """The first echo after a miss logs the recovery and deletes the repair issue.""" + handler = _handler("F461") + frames: asyncio.Queue = asyncio.Queue() + sessions = [] + + def new_session(**_kwargs): + sessions.append(_silent_session(frames)) + return sessions[-1] + + def send_and_echo(message): + if len(sessions) >= 2: + frames.put_nowait(OWNMessage.parse(PROBE_ECHO)) + return _written_now() + + probe_send = MagicMock(side_effect=send_and_echo) + with ( + patch("custom_components.myhome.gateway.OWNEventSession", side_effect=new_session), + patch.object(handler._command_pool, "send_status_request_nowait", probe_send), + patch("custom_components.myhome.repairs.async_create_event_session_silent_issue"), + patch("custom_components.myhome.repairs.async_delete_event_session_silent_issue") as delete_issue, + caplog.at_level(logging.INFO, logger="custom_components.myhome"), + ): + await _run_listener_until(handler, lambda: delete_issue.called) + + assert len(sessions) == 2 + delete_issue.assert_called_with(handler.hass, "entry_probe") + assert handler._event_runner._probe_misses == 0 + assert any("answers the liveness probe again" in r.getMessage() for r in caplog.records) + + +async def test_event_session_silent_repair_is_created_and_deleted(hass): + """The repair issue carries the gateway name and miss count, and clears on delete.""" + from homeassistant.helpers import issue_registry as ir + + from custom_components.myhome.repairs import ( + async_create_event_session_silent_issue, + async_delete_event_session_silent_issue, + ) + + async_create_event_session_silent_issue(hass, "entry_probe", "F461 Gateway", 3) + issue = ir.async_get(hass).async_get_issue("myhome", "event_session_silent_entry_probe") + assert issue is not None + assert issue.translation_key == "event_session_silent" + assert issue.translation_placeholders == {"gateway": "F461 Gateway", "misses": "3"} + + async_delete_event_session_silent_issue(hass, "entry_probe") + assert ir.async_get(hass).async_get_issue("myhome", "event_session_silent_entry_probe") is None + + +def test_nowait_status_request_follows_the_delegated_and_standby_routing(): + """The non-blocking path routes like send_status_request: delegate, then standby, then own pool.""" + handler = _handler("F461") + command = OWNMessage.parse("*#4*1##") + delegate = MagicMock() + standby = MagicMock() + + with patch.object(handler, "_delegated_target", return_value=delegate): + assert handler.send_status_request_nowait(command) is delegate.send_status_request_nowait.return_value + with ( + patch.object(handler, "_delegated_target", return_value=None), + patch.object(handler, "_failover_target", return_value=standby), + ): + assert handler.send_status_request_nowait(command) is standby.send_status_request_nowait.return_value + delegate.send_status_request_nowait.assert_called_once_with(command) + standby.send_status_request_nowait.assert_called_once_with(command) + + +def test_a_seen_echo_outweighs_a_profile_flag_and_the_table(): + """Once this gateway has echoed the probe, a missing echo is a verdict whatever the profile says.""" + handler = _handler("MH202") + handler._event_runner._probe_echo_seen = True + profile = MagicMock() + profile.echoes_event_session_probe = False + with patch.object(type(handler), "profile", new_callable=lambda: property(lambda self: profile)): + assert handler._event_runner._probe_echo_expected() is True + + +@pytest.mark.parametrize( + ("frame", "expected"), + [ + ("*#13**0*20*42*07*001##", True), # parsed: OWNGatewayEvent, dimension 0 + ("*#13**1*06*24*10*2026##", False), # parsed: OWNGatewayEvent, the date (dimension 1) + ("*1*1*12##", False), # a light event + ], +) +def test_time_echo_is_matched_on_the_parsed_gateway_event(frame, expected): + """The echo is the WHO 13 dimension 0 reply as OWNd parses it, not a raw-string prefix.""" + message = OWNMessage.parse(frame) + assert message is not None + assert gateway_sessions._is_time_echo(message) is expected + + +def test_time_echo_falls_back_to_the_raw_string(): + """A reply the pinned OWNd hands back unparsed still counts by its `*#13**0*` prefix.""" + assert gateway_sessions._is_time_echo("*#13**0*20*42*07*0530##") is True + assert gateway_sessions._is_time_echo("*#13**27*70#52#54#49##") is False + diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 11f17a4e6..4acdd38ce 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -738,6 +738,8 @@ async def test_listening_loop_other_events(gateway_handler): mock_session_class.return_value = mock_session gateway_handler.send_status_request = AsyncMock() + # Event-triggered zone polls are queued without blocking the read loop + gateway_handler.send_status_request_nowait = MagicMock() try: await gateway_handler.listening_loop() @@ -774,7 +776,8 @@ async def test_listening_loop_other_events(gateway_handler): "message": str(msg_alarm), }, ) - assert gateway_handler.send_status_request.call_count >= 2 + assert gateway_handler.send_status_request_nowait.call_count >= 2 + gateway_handler.send_status_request.assert_not_called() diff --git a/tests/test_issue_674_f455_climate_trace.py b/tests/test_issue_674_f455_climate_trace.py index 016b1120b..717a83dda 100644 --- a/tests/test_issue_674_f455_climate_trace.py +++ b/tests/test_issue_674_f455_climate_trace.py @@ -273,5 +273,6 @@ async def test_gateway_events_skips_zone_status_query_on_physical_gateways(hass: dispatcher_mhs1._is_active_for_who = lambda who: True # type: ignore[method-assign] await dispatcher_mhs1.process_message(msg) - handler_mhs1.send_status_request.assert_called_once() + # Queued without blocking the event read loop + handler_mhs1.send_status_request_nowait.assert_called_once()