From 21440717dd262b867c1d3e294b23583cd3f7fc7f Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:57:31 +0200 Subject: [PATCH 1/2] feat(connection): add typed exceptions, fast send reconnect, and API hardening --- OWNd/__init__.py | 20 +++ OWNd/connection/__init__.py | 141 +++------------------ OWNd/connection/command_session.py | 169 +++++++++++++++++++++++--- OWNd/connection/event_session.py | 64 ++++++++-- OWNd/connection/session.py | 185 ++++++++++++++++++++++++---- OWNd/exceptions.py | 65 ++++++++++ OWNd/message/gateway.py | 28 ++++- OWNd/transport/base.py | 2 +- OWNd/transport/serial.py | 2 +- OWNd/transport/tcp.py | 2 +- tests/golden/README.md | 2 - tests/test_connection.py | 189 ++++++++++++++++++++++++++++- tests/test_connection_extended.py | 26 +--- tests/test_connection_faults.py | 10 +- tests/test_connection_proxy.py | 153 ----------------------- tests/test_message_exhaustive.py | 4 + tests/typing/consumer_send.py | 26 +++- tools/golden/MASSI_OUTREACH.md | 54 --------- tools/golden/MCP_NOTES.md | 60 --------- 19 files changed, 718 insertions(+), 484 deletions(-) create mode 100644 OWNd/exceptions.py delete mode 100644 tests/test_connection_proxy.py delete mode 100644 tools/golden/MASSI_OUTREACH.md delete mode 100644 tools/golden/MCP_NOTES.md diff --git a/OWNd/__init__.py b/OWNd/__init__.py index 4bdf1937..946826ef 100755 --- a/OWNd/__init__.py +++ b/OWNd/__init__.py @@ -1,2 +1,22 @@ """ OWNd - an OpenWebNet daemon """ # pylint: disable=invalid-name + +from .exceptions import ( + OWNAuthError, + OWNConnectionError, + OWNError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, +) + __version__ = "2.0.0-b10" + +__all__ = [ + "__version__", + "OWNAuthError", + "OWNConnectionError", + "OWNError", + "OWNNackError", + "OWNProtocolError", + "OWNTimeoutError", +] diff --git a/OWNd/connection/__init__.py b/OWNd/connection/__init__.py index 900076f3..18a74ed0 100644 --- a/OWNd/connection/__init__.py +++ b/OWNd/connection/__init__.py @@ -2,27 +2,6 @@ from __future__ import annotations -# Standard library and utility imports retained exclusively for backwards compatibility -# with consumer test suites and legacy mock/patch targets (e.g., patch("OWNd.connection.time"), -# patch("OWNd.connection.socket"), patch("OWNd.connection.asyncio")). -# While not directly referenced in this module's runtime logic, _ConnectionModule forwards -# patched attributes to underlying submodules. DO NOT REMOVE or "clean up" these imports. -import asyncio -import contextlib -import hashlib -import hmac -import logging -import secrets -import socket -import string -import sys -import time -from collections.abc import Callable, Mapping -from types import ModuleType -from typing import Any -from urllib.parse import urlparse - -from . import auth, command_session, event_session, gateway, session from .auth import ( calculate_open_password, decode_hmac_response, @@ -30,7 +9,7 @@ hex_string_to_int_string, int_string_to_hex_string, ) -from .command_session import OWNCommandSession +from .command_session import OWNCommandSession, SEND_RECONNECT_COOLDOWN from .event_session import OWNEventSession from .gateway import OWNGateway, _first_scalar from .session import ( @@ -54,9 +33,18 @@ TCP_KEEPALIVE_CNT, TCP_KEEPALIVE_IDLE, TCP_KEEPALIVE_INTVL, + _AUTH_NEGOTIATION_ERRORS, _FATAL_NEGOTIATION_ERRORS, ) from ..discovery import find_gateways, get_gateway, get_port +from ..exceptions import ( + OWNAuthError, + OWNConnectionError, + OWNError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, +) from ..message import OWNMessage, OWNSignaling from ..profiles import GatewayProfile, get_gateway_profile @@ -67,6 +55,13 @@ "OWNSession", "OWNEventSession", "OWNCommandSession", + # Exceptions + "OWNError", + "OWNConnectionError", + "OWNTimeoutError", + "OWNAuthError", + "OWNNackError", + "OWNProtocolError", # Constants "NEGOTIATION_TIMEOUT", "NEGOTIATION_TOTAL_TIMEOUT", @@ -75,6 +70,7 @@ "COMMAND_RESPONSE_MAX_FRAMES", "CONNECT_TIMEOUT", "MAX_CONNECT_ATTEMPTS", + "SEND_RECONNECT_COOLDOWN", "KEEPALIVE_FRAME", "KEEPALIVE_INTERVAL", "EVENT_KEEPALIVE_FRAME", @@ -82,6 +78,7 @@ "TCP_KEEPALIVE_IDLE", "TCP_KEEPALIVE_INTVL", "TCP_KEEPALIVE_CNT", + "_AUTH_NEGOTIATION_ERRORS", "_FATAL_NEGOTIATION_ERRORS", "RECONNECT_PAUSE", "RECONNECT_PAUSE_FATAL", @@ -105,103 +102,3 @@ "hex_string_to_int_string", "int_string_to_hex_string", ] - - -_FORWARD_TARGETS: dict[str, tuple[str, ...]] = { - "time": ("session",), - "socket": ("session",), - "asyncio": ("session",), - "secrets": ("session",), - "hashlib": ("auth",), - "hmac": ("session",), - "string": ("session",), - "logging": ("session", "gateway"), - "urlparse": ("gateway", "session"), - "find_gateways": ("gateway",), - "get_gateway": ("gateway",), - "get_port": ("gateway",), - "GatewayProfile": ("gateway", "session"), - "get_gateway_profile": ("gateway", "session"), -} - - -class _ConnectionModule(ModuleType): - """Proxy module synchronizing mocked attributes with submodules. - - Ensures 100% backward compatibility when test suites or consumers patch - attributes directly on ``OWNd.connection`` (e.g. ``time``, ``find_gateways``, - ``NEGOTIATION_MAX_FRAMES``). - """ - - def _get_submodules(self) -> list[ModuleType]: - """Dynamically resolve all loaded connection submodules.""" - pkg = self.__name__ - return [ - mod - for name, mod in list(sys.modules.items()) - if name.startswith(f"{pkg}.") - and isinstance(mod, ModuleType) - and mod.__name__ != pkg - ] - - def __getattr__(self, name: str) -> Any: - for mod in self._get_submodules(): - if hasattr(mod, name): - return getattr(mod, name) - raise AttributeError(f"module '{self.__name__}' has no attribute '{name}'") - - def __setattr__(self, name: str, value: Any) -> None: - if name.startswith("__") and name.endswith("__"): - super().__setattr__(name, value) - return - - if name in _FORWARD_TARGETS: - pkg = self.__name__ - for sub_name in _FORWARD_TARGETS[name]: - target_mod = sys.modules.get(f"{pkg}.{sub_name}") - if target_mod is None or not hasattr(target_mod, name): - raise AttributeError( - f"Patched attribute '{name}' missing from expected target submodule '{sub_name}'" - ) - - all_attrs = getattr(self, "__all__", ()) - submodules = self._get_submodules() - if name in all_attrs and submodules: - if not any(hasattr(mod, name) for mod in submodules): - raise AttributeError( - f"Patched attribute '{name}' in __all__ is missing from all connection submodules" - ) - - super().__setattr__(name, value) - for mod in submodules: - if hasattr(mod, name): - setattr(mod, name, value) - - def __delattr__(self, name: str) -> None: - if name.startswith("__") and name.endswith("__"): - super().__delattr__(name) - return - - deleted = False - with contextlib.suppress(AttributeError): - super().__delattr__(name) - deleted = True - - for mod in self._get_submodules(): - if hasattr(mod, name): - with contextlib.suppress(AttributeError): - delattr(mod, name) - deleted = True - - if not deleted: - raise AttributeError(f"module '{self.__name__}' has no attribute '{name}'") - - def __dir__(self) -> list[str]: - attrs = set(super().__dir__()) - for mod in self._get_submodules(): - attrs.update(dir(mod)) - return sorted(attrs) - - -sys.modules[__name__].__class__ = _ConnectionModule - diff --git a/OWNd/connection/command_session.py b/OWNd/connection/command_session.py index 1245f2be..ce793cef 100644 --- a/OWNd/connection/command_session.py +++ b/OWNd/connection/command_session.py @@ -5,6 +5,7 @@ import asyncio import contextlib import logging +import time from collections.abc import Callable from typing import Any @@ -16,32 +17,66 @@ KEEPALIVE_INTERVAL, OWNSession, ) +from ..exceptions import ( + OWNAuthError, + OWNConnectionError, + OWNError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, +) from ..message import OWNMessage, OWNSignaling +# Cooldown window (seconds) after an in-send() reconnect attempt fails, so +# concurrent callers queued behind _send_lock fail fast instead of each +# repeating a socket connect attempt while the gateway is unreachable. +SEND_RECONNECT_COOLDOWN = 2.0 + + class OWNCommandSession(OWNSession): def __init__( self, gateway: OWNGateway | None = None, logger: logging.Logger | None = None, on_state_change: Callable[[bool], None] | None = None, + on_auth_error: Callable[[OWNAuthError], Any] | None = None, ): super().__init__( gateway=gateway, connection_type="command", logger=logger, on_state_change=on_state_change, + on_auth_error=on_auth_error, ) # Serializes concurrent senders sharing this session (e.g. the # keepalive loop and a regular command): interleaved write/read pairs # on the same stream would steal each other's acknowledgements. self._send_lock = asyncio.Lock() + self._last_reconnect_failure: float | None = None + + async def connect( + self, *, raise_on_error: bool = False + ) -> dict[str, Any] | None: + if self._connect_max_attempts_override is None: + self._last_reconnect_failure = None + result = await super().connect(raise_on_error=raise_on_error) + if result is not None and result.get("Success", False): + self._last_reconnect_failure = None + return result @classmethod async def send_to_gateway( - cls, message: str | OWNMessage, gateway: OWNGateway + cls, + message: str | OWNMessage, + gateway: OWNGateway, + *, + raise_on_error: bool = False, ) -> list[OWNMessage | str] | bool | None: connection = cls(gateway) try: + if raise_on_error: + await connection.connect(raise_on_error=True) + return await connection.send(message, raise_on_error=True) await connection.connect() return await connection.send(message) finally: @@ -50,10 +85,12 @@ async def send_to_gateway( @classmethod async def connect_to_gateway( - cls, gateway: OWNGateway + cls, gateway: OWNGateway, *, raise_on_error: bool = False ) -> dict[str, Any] | None: connection = cls(gateway) try: + if raise_on_error: + return await connection.connect(raise_on_error=True) return await connection.connect() finally: with contextlib.suppress(Exception): @@ -159,7 +196,11 @@ async def _read_signaling_response(self) -> OWNSignaling: return signaling async def send( - self, message: str | OWNMessage, is_status_request: bool = False + self, + message: str | OWNMessage, + is_status_request: bool = False, + *, + raise_on_error: bool = False, ) -> list[OWNMessage | str] | bool | None: """Send the attached message on an existing 'command' connection, actively reconnecting it if it had been reset. @@ -178,11 +219,18 @@ async def send( the call stack nor loop forever. """ async with self._send_lock: - return await self._locked_send(message, is_status_request) + return await self._locked_send( + message, is_status_request, raise_on_error=raise_on_error + ) async def _locked_send( - self, message: str | OWNMessage, is_status_request: bool = False + self, + message: str | OWNMessage, + is_status_request: bool = False, + *, + raise_on_error: bool = False, ) -> list[OWNMessage | str] | bool | None: + self._last_error = None msg_obj = ( message if isinstance(message, OWNMessage) @@ -194,11 +242,19 @@ async def _locked_send( and not isinstance(msg_obj, OWNSignaling) ) if not is_valid_frame: + err: OWNError = OWNProtocolError( + f"Invalid OpenWebNet frame `{message}`", + frame=str(message), + reason="invalid_frame", + ) + self._last_error = err self._logger.error( "%s Refusing to send invalid OpenWebNet frame `%s`.", self._log_id, message, ) + if raise_on_error: + raise err return None # One retry is enough for a command NACK or for a connection that was @@ -209,28 +265,67 @@ async def _locked_send( # gateway may have executed it even if its acknowledgement was lost. # Status requests are idempotent and can be retried safely after a # transport reset. + is_idempotent_query = is_status_request and ( + getattr(msg_obj, "_message_type", None) + in ("STATUS_REQUEST", "DIMENSION_REQUEST") + ) max_attempts = 2 for attempt in range(1, max_attempts + 1): # After an outage the previous connect()/reconnect may have given up - # and left the writer at None; rebuild the session here so commands - # resume automatically when the gateway comes back, instead of - # crashing forever on `NoneType.write`. + # and left the writer at None; rebuild the session here with a + # single fast connect attempt (and a short cooldown on failure) so + # queued callers behind _send_lock never stall for a full 5-attempt + # exponential backoff loop. if self._stream_reader is None or self._stream_writer is None: - result = await self.connect() + now = time.monotonic() + if ( + self._last_reconnect_failure is not None + and (now - self._last_reconnect_failure) + < SEND_RECONNECT_COOLDOWN + ): + err = self._last_error or OWNConnectionError( + f"Command session unavailable; message `{message}` not sent", + reason="connection_unavailable", + ) + self._last_error = err + self._logger.warning( + "%s Command session unavailable; message `%s` not sent.", + self._log_id, + message, + ) + if raise_on_error: + raise err + return None + + self._connect_max_attempts_override = 1 + try: + result = await self.connect() + finally: + self._connect_max_attempts_override = None + if ( result is None or not result.get("Success", False) or self._stream_reader is None or self._stream_writer is None ): + self._last_reconnect_failure = time.monotonic() + err = self._last_error or OWNConnectionError( + f"Command session unavailable; message `{message}` not sent", + reason="connection_unavailable", + ) await self.close() + self._last_error = err self._logger.warning( "%s Command session unavailable; message `%s` not sent.", self._log_id, message, ) + if raise_on_error: + raise err return None + self._last_reconnect_failure = None written = False try: @@ -241,11 +336,11 @@ async def _locked_send( resulting_message, collected = await self._read_command_response() if resulting_message.is_ack(): - log_message = "%s Message `%s` was successfully sent." - if not is_status_request: - self._logger.info(log_message, self._log_id, message) - else: - self._logger.debug(log_message, self._log_id, message) + self._logger.debug( + "%s Message `%s` was successfully sent.", + self._log_id, + message, + ) return collected or True if resulting_message.is_nack(): @@ -265,6 +360,14 @@ async def _locked_send( self._log_id, message, ) + err = OWNNackError( + f"Gateway rejected message `{message}` with NACK", + frame=str(message), + responses=list(collected), + ) + self._last_error = err + if raise_on_error: + raise err return None self._logger.error( "%s Could not send message `%s`. Retrying (%d)...", @@ -280,19 +383,36 @@ async def _locked_send( resulting_message, message, ) + err = OWNProtocolError( + f"Unexpected response `{resulting_message}` to message `{message}`", + frame=str(resulting_message), + reason="unexpected_response", + ) + self._last_error = err + if raise_on_error: + raise err return None except asyncio.CancelledError: await self.close() raise - except TimeoutError: + except OWNError: + raise + except TimeoutError as error: await self.close() + err = OWNTimeoutError( + f"Timed out awaiting the complete response for `{message}`", + reason="command_timeout", + ) + self._last_error = err self._logger.warning( "%s Timed out awaiting the complete response for `%s`; " "command session closed.", self._log_id, message, ) + if raise_on_error: + raise err from error return None except ( ConnectionResetError, @@ -304,7 +424,7 @@ async def _locked_send( # state yet: a retry that succeeds is an internal recycle and # must not flap the consumer (offline, then online again). await self._close_streams() - if attempt < max_attempts and (is_status_request or not written): + if attempt < max_attempts and (is_idempotent_query or not written): self._logger.debug( "%s Command session connection lost (%s), retrying once...", self._log_id, @@ -313,19 +433,32 @@ async def _locked_send( continue # No retry: this is a definitive loss, and the consumer hears it. self._set_connected(False) + err = OWNConnectionError( + f"Connection lost before acknowledgement of `{message}` ({error})", + reason="connection_lost", + ) + self._last_error = err self._logger.warning( "%s Connection lost before acknowledgement of `%s`; " "message will not be replayed.", self._log_id, message, ) + if raise_on_error: + raise err from error return None - except Exception: # pylint: disable=broad-except + except Exception as error: # pylint: disable=broad-except await self.close() + err = OWNConnectionError( + f"Command session crashed ({error})", + reason="session_crash", + ) + self._last_error = err self._logger.exception( "%s Command session crashed.", self._log_id ) + if raise_on_error: + raise err from error return None return None # pragma: no cover - diff --git a/OWNd/connection/event_session.py b/OWNd/connection/event_session.py index 9da3038d..772a72fb 100644 --- a/OWNd/connection/event_session.py +++ b/OWNd/connection/event_session.py @@ -19,10 +19,13 @@ OWNSession, RECONNECT_PAUSE, RECONNECT_PAUSE_FATAL, + _AUTH_NEGOTIATION_ERRORS, _FATAL_NEGOTIATION_ERRORS, ) +from ..exceptions import OWNAuthError from ..message import OWNMessage + class OWNEventSession(OWNSession): def __init__( self, @@ -30,16 +33,20 @@ def __init__( logger: logging.Logger | None = None, inactivity_timeout: float | None = EVENT_INACTIVITY_TIMEOUT, on_state_change: Callable[[bool], None] | None = None, + on_auth_error: Callable[[OWNAuthError], Any] | None = None, + raise_on_auth_error: bool = False, ): super().__init__( gateway=gateway, connection_type="event", logger=logger, on_state_change=on_state_change, + on_auth_error=on_auth_error, ) # Passive watchdog: reconnect if no frame arrives for this long. # Set to None to disable (pure blocking read, never times out). self._inactivity_timeout = inactivity_timeout + self._raise_on_auth_error = raise_on_auth_error # The event session is the long-lived monitored connection: let the OS # actively probe it so a silent outage is detected in ~60s. self._tcp_keepalive = True @@ -50,9 +57,11 @@ def __init__( self._recent_drops: list[float] = [] self._last_drop_warning: float | None = None - async def connect(self) -> dict[str, Any] | None: + async def connect( + self, *, raise_on_error: bool = False + ) -> dict[str, Any] | None: await self._stop_keepalive() - result = await super().connect() + result = await super().connect(raise_on_error=raise_on_error) if result is not None and result.get("Success"): self._start_keepalive() return result @@ -98,16 +107,45 @@ async def _close_streams(self) -> None: @classmethod async def connect_to_gateway( - cls, gateway: OWNGateway + cls, gateway: OWNGateway, *, raise_on_error: bool = False ) -> dict[str, Any] | None: connection = cls(gateway) try: + if raise_on_error: + return await connection.connect(raise_on_error=True) return await connection.connect() finally: with contextlib.suppress(Exception): await connection.close() - async def get_next(self) -> OWNMessage | str | None: + def _maybe_raise_auth_error( + self, + result: dict[str, Any] | None, + raise_on_auth_error: bool | None, + ) -> None: + should_raise = ( + self._raise_on_auth_error + if raise_on_auth_error is None + else raise_on_auth_error + ) + if not should_raise: + return + if isinstance(self._last_error, OWNAuthError): + raise self._last_error + if ( + result is not None + and result.get("Message") in _AUTH_NEGOTIATION_ERRORS + ): + reason = str(result.get("Message")) + err = OWNAuthError( + f"Gateway authentication failed ({reason})", reason=reason + ) + self._last_error = err + raise err + + async def get_next( + self, *, raise_on_auth_error: bool | None = None + ) -> OWNMessage | str | None: """Acts as an entry point to read messages on the event bus. It will read one frame and return it as an OWNMessage object. @@ -127,6 +165,7 @@ async def get_next(self) -> OWNMessage | str | None: ) result = await self._reconnect() if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) # Still no stream: connect() gave up or failed fatally and # returned IMMEDIATELY (its internal back-off only covers the # transient retries). Pause here, otherwise the caller's read @@ -157,14 +196,18 @@ async def get_next(self) -> OWNMessage | str | None: self._log_id, self._inactivity_timeout, ) - await self._reconnect() + result = await self._reconnect() + if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) return None except asyncio.LimitOverrunError: self._logger.warning( "%s Received oversized or garbage frame, dropping connection and reconnecting...", self._log_id, ) - await self._reconnect() + result = await self._reconnect() + if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) return None except ( asyncio.IncompleteReadError, @@ -200,13 +243,17 @@ async def get_next(self) -> OWNMessage | str | None: self._logger.debug( "%s Event connection lost, reconnecting...", self._log_id ) - await self._reconnect() + result = await self._reconnect() + if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) return None except Exception: # pylint: disable=broad-except self._logger.exception( "%s Event session crashed, reconnecting...", self._log_id ) - await self._reconnect() + result = await self._reconnect() + if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) return None # A frame was read successfully: from here on, any failure is a @@ -225,4 +272,3 @@ async def get_next(self) -> OWNMessage | str | None: error, ) return _decoded_data - diff --git a/OWNd/connection/session.py b/OWNd/connection/session.py index 420a02a8..952fec28 100644 --- a/OWNd/connection/session.py +++ b/OWNd/connection/session.py @@ -23,6 +23,14 @@ ) from .gateway import OWNGateway, _first_scalar from ..discovery import get_port +from ..exceptions import ( + OWNAuthError, + OWNConnectionError, + OWNError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, +) from ..message import OWNMessage, OWNSignaling from ..profiles import GatewayProfile, get_gateway_profile @@ -60,6 +68,8 @@ TCP_KEEPALIVE_IDLE = 30 # start probing after 30s of silence TCP_KEEPALIVE_INTVL = 10 # probe every 10s TCP_KEEPALIVE_CNT = 3 # declare dead after 3 missed probes (~60s total) +# Authentication negotiation failures (missing/wrong/unsupported password). +_AUTH_NEGOTIATION_ERRORS = frozenset({"password_required", "password_error"}) # Negotiation failures that will never succeed on a retry: don't loop on them. _FATAL_NEGOTIATION_ERRORS = frozenset( {"password_required", "password_error", "negotiation_error"} @@ -84,6 +94,29 @@ DROP_WARNING_INTERVAL = 3600 # 1 hour +def _negotiation_error_to_exception(reason: str | None) -> OWNError: + """Map a negotiation error code to a typed OWNError subclass.""" + if reason in _AUTH_NEGOTIATION_ERRORS: + return OWNAuthError( + f"Gateway authentication failed ({reason})", reason=reason + ) + if reason == "negotiation_timeout": + return OWNTimeoutError( + "Timed out negotiating OpenWebNet session", reason=reason + ) + if reason == "negotiation_refused": + return OWNNackError( + "Gateway refused session negotiation with NACK", reason=reason + ) + if reason in ("negotiation_error", "negotiation_failed"): + return OWNProtocolError( + f"OpenWebNet negotiation protocol error ({reason})", reason=reason + ) + return OWNConnectionError( + f"Gateway connection failed ({reason or 'unknown'})", reason=reason + ) + + class OWNSession: """Connection to OpenWebNet gateway""" @@ -95,6 +128,7 @@ def __init__( connection_type: str = "test", logger: logging.Logger | None = None, on_state_change: Callable[[bool], None] | None = None, + on_auth_error: Callable[[OWNAuthError], Any] | None = None, ): """Initialize the class Arguments: @@ -105,6 +139,9 @@ def __init__( connection comes up / goes down. Lets a consumer (e.g. the Home Assistant integration) flip entity availability instantly instead of waiting for the next keepalive. + on_auth_error: optional callback invoked with an ``OWNAuthError`` when + session authentication fails (e.g. password changed on the gateway + at runtime so Home Assistant can trigger a reauthentication flow). """ self._gateway = gateway @@ -114,6 +151,9 @@ def __init__( # instead of crashing on `None.warning(...)` and masking the real error. self._logger = logger if logger is not None else logging.getLogger(__name__) self._on_state_change = on_state_change + self._on_auth_error = on_auth_error + self._last_error: OWNError | None = None + self._connect_max_attempts_override: int | None = None self._connected = False # Enable OS-level TCP keepalive on the socket (event session only). self._tcp_keepalive = False @@ -122,6 +162,33 @@ def __init__( self._stream_reader: asyncio.StreamReader | None = None self._stream_writer: asyncio.StreamWriter | None = None + @property + def last_error(self) -> OWNError | None: + """The typed ``OWNError`` from the most recent failed operation, or ``None``.""" + return self._last_error + + @property + def on_auth_error(self) -> Callable[[OWNAuthError], Any] | None: + """Optional callback invoked when gateway authentication is rejected.""" + return self._on_auth_error + + @on_auth_error.setter + def on_auth_error( + self, callback: Callable[[OWNAuthError], Any] | None + ) -> None: + self._on_auth_error = callback + + def _notify_auth_error(self, error: OWNAuthError) -> None: + """Invoke the consumer's auth-failure callback without letting it crash the session.""" + if self._on_auth_error is not None: + try: + self._on_auth_error(error) + except Exception: # noqa: BLE001 + if self._logger is not None: + self._logger.exception( + "%s on_auth_error callback raised.", self._log_id + ) + @property def is_connected(self) -> bool: """True once a session has been negotiated and not since lost.""" @@ -246,12 +313,19 @@ def connection_type(self, connection_type: str) -> None: self._type = connection_type.lower() @classmethod - async def test_gateway(cls, gateway: OWNGateway) -> dict[str, Any]: + async def test_gateway( + cls, gateway: OWNGateway, *, raise_on_error: bool = False + ) -> dict[str, Any]: connection = cls(gateway) + if raise_on_error: + return await connection.test_connection(raise_on_error=True) return await connection.test_connection() - async def test_connection(self) -> dict[str, Any]: + async def test_connection( + self, *, raise_on_error: bool = False + ) -> dict[str, Any]: assert self._gateway is not None + self._last_error = None retry_count = 0 retry_timer = 1 @@ -262,6 +336,13 @@ async def test_connection(self) -> dict[str, Any]: "%s Test session connection still refused after 3 attempts.", self._log_id, ) + err: OWNError = OWNConnectionError( + "Test session connection still refused after 3 attempts", + reason="connection_error", + ) + self._last_error = err + if raise_on_error: + raise err return {"Success": False, "Message": "connection_error"} ( self._stream_reader, @@ -286,13 +367,28 @@ async def test_connection(self) -> dict[str, Any]: try: try: - return await self._negotiate() + result = await self._negotiate() + if not result.get("Success"): + err = _negotiation_error_to_exception(result.get("Message")) + self._last_error = err + if isinstance(err, OWNAuthError): + self._notify_auth_error(err) + if raise_on_error: + raise err + return result except ConnectionResetError: self._logger.error( "%s Negotiation reset while opening %s session. Wait 60 seconds before retrying.", self._log_id, self._type, ) + err = OWNConnectionError( + "Negotiation reset while opening session", + reason="password_retry", + ) + self._last_error = err + if raise_on_error: + raise err from None return {"Success": False, "Message": "password_retry"} except ( asyncio.IncompleteReadError, @@ -310,6 +406,20 @@ async def test_connection(self) -> dict[str, Any]: self._type, error, ) + err = ( + OWNTimeoutError( + f"Negotiation timed out ({error})", + reason="connection_error", + ) + if isinstance(error, TimeoutError) + else OWNConnectionError( + f"Negotiation failed ({error})", + reason="connection_error", + ) + ) + self._last_error = err + if raise_on_error: + raise err from error return {"Success": False, "Message": "connection_error"} finally: # Test sessions are always temporary. Release the descriptor on @@ -317,11 +427,18 @@ async def test_connection(self) -> dict[str, Any]: with contextlib.suppress(Exception): await self.close() - async def connect(self) -> dict[str, Any] | None: + async def connect( + self, *, raise_on_error: bool = False + ) -> dict[str, Any] | None: assert self._gateway is not None self._logger.debug("%s Opening %s session.", self._log_id, self._type) + self._last_error = None + max_attempts = self._connect_max_attempts_override or MAX_CONNECT_ATTEMPTS retry_count = 0 + last_attempt_error: OWNError = OWNConnectionError( + "Session could not be established", reason="connection_error" + ) while True: try: @@ -336,36 +453,57 @@ async def connect(self) -> dict[str, Any] | None: self._apply_tcp_keepalive() result = await self._negotiate() if result.get("Success"): + self._last_error = None self._set_connected(True) return result # Negotiation completed but was rejected. - if result.get("Message") in _FATAL_NEGOTIATION_ERRORS: + msg = result.get("Message") + last_attempt_error = _negotiation_error_to_exception(msg) + if msg in _FATAL_NEGOTIATION_ERRORS: self._logger.error( "%s %s session negotiation failed (%s); giving up.", self._log_id, self._type.capitalize(), - result.get("Message"), + msg, ) + self._last_error = last_attempt_error + if isinstance(last_attempt_error, OWNAuthError): + self._notify_auth_error(last_attempt_error) self._set_connected(False) # The TCP connection survived the rejected negotiation: # release it, it will not be used. with contextlib.suppress(Exception): await self._close_streams() + if raise_on_error: + raise last_attempt_error return result - reason = f"negotiation failed ({result.get('Message')})" + reason = f"negotiation failed ({msg})" wait = max(1, retry_count * 2) except ConnectionResetError: reason, wait = "connection reset", 60 + last_attempt_error = OWNConnectionError( + "Connection reset by peer", reason="connection_reset" + ) except (ConnectionRefusedError, asyncio.IncompleteReadError): reason, wait = "connection refused", max(1, retry_count * 2) + last_attempt_error = OWNConnectionError( + "Connection refused", reason="connection_refused" + ) except TimeoutError: reason, wait = ( f"connect timed out ({CONNECT_TIMEOUT}s)", max(1, retry_count * 2), ) + last_attempt_error = OWNTimeoutError( + f"Connect timed out ({CONNECT_TIMEOUT}s)", + reason="connect_timeout", + ) except OSError as error: # Host unreachable, no route, DNS failure, etc. reason, wait = f"network error ({error})", max(1, retry_count * 2) + last_attempt_error = OWNConnectionError( + f"Network error ({error})", reason="network_error" + ) retry_count += 1 # A failed attempt can leave a half-open socket behind (e.g. TCP @@ -375,15 +513,19 @@ async def connect(self) -> dict[str, Any] | None: # below, so a retry that succeeds never flaps the consumer. with contextlib.suppress(Exception): await self._close_streams() - if retry_count >= MAX_CONNECT_ATTEMPTS: - self._logger.warning( - "%s %s session could not be established after %d attempts; " - "will retry.", - self._log_id, - self._type.capitalize(), - MAX_CONNECT_ATTEMPTS, - ) + if retry_count >= max_attempts: + self._last_error = last_attempt_error + if max_attempts > 1: + self._logger.warning( + "%s %s session could not be established after %d attempts; " + "will retry.", + self._log_id, + self._type.capitalize(), + max_attempts, + ) self._set_connected(False) + if raise_on_error: + raise last_attempt_error return None self._logger.warning( "%s %s session: %s. Retrying in %ss (attempt %d/%d).", @@ -392,7 +534,7 @@ async def connect(self) -> dict[str, Any] | None: reason, wait, retry_count, - MAX_CONNECT_ATTEMPTS, + max_attempts, ) await asyncio.sleep(wait) @@ -428,13 +570,14 @@ async def _close_streams(self) -> None: # May be invoked on an empty instance, or on an already-broken socket: # be robust against Nones and against wait_closed() re-raising. - if self._stream_writer is not None: - self._stream_writer.close() - # The peer may already be gone; we only need it marked closed. - with contextlib.suppress(OSError, asyncio.IncompleteReadError): - await self._stream_writer.wait_closed() + writer = self._stream_writer self._stream_reader = None self._stream_writer = None + if writer is not None: + writer.close() + # The peer may already be gone; we only need it marked closed. + with contextlib.suppress(OSError, asyncio.IncompleteReadError): + await writer.wait_closed() if self._gateway is not None: self._logger.debug( "%s %s session closed.", self._log_id, self._type.capitalize() diff --git a/OWNd/exceptions.py b/OWNd/exceptions.py new file mode 100644 index 00000000..ec4bbda4 --- /dev/null +++ b/OWNd/exceptions.py @@ -0,0 +1,65 @@ +"""Typed exceptions for OpenWebNet connection, authentication, and protocol errors.""" + +from __future__ import annotations + +from typing import Any + + +class OWNError(Exception): + """Base exception for all OWNd errors.""" + + def __init__(self, message: str = "OpenWebNet error", *, reason: str | None = None) -> None: + super().__init__(message) + self.reason = reason + + +class OWNConnectionError(OWNError): + """Socket or transport connection failure (refused, reset, EOF, or unavailable).""" + + +class OWNTimeoutError(OWNConnectionError): + """Timed out during TCP connection, session negotiation, or command response.""" + + +class OWNAuthError(OWNError): + """Gateway authentication failed (missing password, invalid password, or rejected digest).""" + + +class OWNNackError(OWNError): + """Gateway rejected a frame or negotiation step with NACK (``*#*0##``).""" + + def __init__( + self, + message: str = "Gateway rejected frame with NACK", + *, + frame: str = "", + responses: list[Any] | None = None, + reason: str | None = "nack", + ) -> None: + super().__init__(message, reason=reason) + self.frame = frame + self.responses: list[Any] = list(responses) if responses is not None else [] + + +class OWNProtocolError(OWNError): + """Invalid outbound OpenWebNet frame or unexpected signaling response.""" + + def __init__( + self, + message: str = "Invalid or unexpected OpenWebNet frame", + *, + frame: str = "", + reason: str | None = "protocol_error", + ) -> None: + super().__init__(message, reason=reason) + self.frame = frame + + +__all__ = [ + "OWNAuthError", + "OWNConnectionError", + "OWNError", + "OWNNackError", + "OWNProtocolError", + "OWNTimeoutError", +] diff --git a/OWNd/message/gateway.py b/OWNd/message/gateway.py index 27eca345..fffe2d3d 100644 --- a/OWNd/message/gateway.py +++ b/OWNd/message/gateway.py @@ -426,9 +426,21 @@ def is_timezone_unconfigured(self) -> bool: """Return True if the gateway reported an unconfigured timezone (sentinel 999).""" return self._raw_timezone == "999" + @staticmethod + def _resolve_tzinfo(time_zone: _dt.tzinfo | str) -> _dt.tzinfo: + if isinstance(time_zone, _dt.tzinfo): + return time_zone + if isinstance(time_zone, str): + return ZoneInfo(time_zone) + raise ValueError( + "time_zone must be a datetime.tzinfo instance or IANA timezone string" + ) + @classmethod - def set_datetime_to_now(cls, time_zone: str) -> OWNGatewayCommand: - timezone = ZoneInfo(time_zone) + def set_datetime_to_now( + cls, time_zone: _dt.tzinfo | str + ) -> OWNGatewayCommand: + timezone = cls._resolve_tzinfo(time_zone) now = _dt.datetime.now(timezone) timezone_offset = ( f"0{now.strftime('%z')[1:3]}" @@ -442,16 +454,20 @@ def set_datetime_to_now(cls, time_zone: str) -> OWNGatewayCommand: return message @classmethod - def set_date_to_today(cls, time_zone: str) -> OWNGatewayCommand: - timezone = ZoneInfo(time_zone) + def set_date_to_today( + cls, time_zone: _dt.tzinfo | str + ) -> OWNGatewayCommand: + timezone = cls._resolve_tzinfo(time_zone) now = _dt.datetime.now(timezone) message = cls(f"*#13**#1*0{now.strftime('%w*%d*%m*%Y##')}") message._human_readable_log = f"Setting gateway date to: {message._date}." return message @classmethod - def set_time_to_now(cls, time_zone: str) -> OWNGatewayCommand: - timezone = ZoneInfo(time_zone) + def set_time_to_now( + cls, time_zone: _dt.tzinfo | str + ) -> OWNGatewayCommand: + timezone = cls._resolve_tzinfo(time_zone) now = _dt.datetime.now(timezone) timezone_offset = ( f"0{now.strftime('%z')[1:3]}" diff --git a/OWNd/transport/base.py b/OWNd/transport/base.py index 36ec952b..e98da9de 100644 --- a/OWNd/transport/base.py +++ b/OWNd/transport/base.py @@ -44,7 +44,7 @@ async def disconnect(self) -> None: @abstractmethod async def send( - self, message: str, is_status_request: bool = False + self, message: str | OWNMessage, is_status_request: bool = False ) -> list[OWNMessage | str] | bool | None: """Send one command and return its response.""" diff --git a/OWNd/transport/serial.py b/OWNd/transport/serial.py index 83287652..b3f97547 100644 --- a/OWNd/transport/serial.py +++ b/OWNd/transport/serial.py @@ -142,7 +142,7 @@ def _process_frame(self, frame: str) -> None: self._pending_frames.append(message) async def send( - self, message: str, is_status_request: bool = False + self, message: str | OWNMessage, is_status_request: bool = False ) -> list[OWNMessage | str] | bool | None: del is_status_request if not self._connected or self._writer is None: diff --git a/OWNd/transport/tcp.py b/OWNd/transport/tcp.py index 3150cf31..7b949458 100644 --- a/OWNd/transport/tcp.py +++ b/OWNd/transport/tcp.py @@ -91,7 +91,7 @@ async def disconnect(self) -> None: await command_session.close() async def send( - self, message: str, is_status_request: bool = False + self, message: str | OWNMessage, is_status_request: bool = False ) -> list[OWNMessage | str] | bool | None: if self._command_session is None: raise RuntimeError("TCP transport is not connected") diff --git a/tests/golden/README.md b/tests/golden/README.md index 1f053819..1ae3bfc2 100644 --- a/tests/golden/README.md +++ b/tests/golden/README.md @@ -55,11 +55,9 @@ tests/golden/ who18_energy.yaml # WHO=18 Energy Management who25_cen_plus.yaml # WHO=25 CEN+ & Dry Contacts tools/golden/ - MCP_NOTES.md # MCP tool capabilities & verification log validate_corpus.py # Corpus schema validator harvest_4j_fixtures.py # Oracle test harvester & divergence detector HARVEST_REPORT.md # Harvest analysis report - MASSI_OUTREACH.md # Community proposal draft for Massi Valla tests/golden/ corpus.json # Zero-dependency standard library runtime cache firmware_oracle.json # Hash-pinned empirical verdict index from own-firmware-oracle diff --git a/tests/test_connection.py b/tests/test_connection.py index 65e6085b..ad77eae3 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -229,7 +229,7 @@ async def stalled_exchange() -> dict: return {"Success": True, "Message": None} session._negotiate_exchange = stalled_exchange - with patch("OWNd.connection.NEGOTIATION_TOTAL_TIMEOUT", 0.01): + with patch("OWNd.connection.session.NEGOTIATION_TOTAL_TIMEOUT", 0.01): result = await session._negotiate() assert result == {"Success": False, "Message": "negotiation_timeout"} @@ -243,7 +243,7 @@ async def test_temporary_test_session_always_closes() -> None: ) with patch( - "OWNd.connection.asyncio.open_connection", + "OWNd.connection.session.asyncio.open_connection", new=AsyncMock(return_value=(asyncio.StreamReader(), writer)), ): result = await session.test_connection() @@ -273,7 +273,7 @@ async def test_command_response_frame_count_is_bounded() -> None: session._read_frame = AsyncMock(return_value="*1*1*12##") with ( - patch("OWNd.connection.COMMAND_RESPONSE_MAX_FRAMES", 3), + patch("OWNd.connection.command_session.COMMAND_RESPONSE_MAX_FRAMES", 3), pytest.raises(TimeoutError), ): await session._read_command_response() @@ -451,7 +451,10 @@ async def test_event_connection_loss_oversized_frame_is_warning() -> None: def _fake_clock(*times: float): """Patch only the module's ``time`` so asyncio's own loop clock is untouched.""" it = iter(times) - return patch("OWNd.connection.time", SimpleNamespace(monotonic=lambda: next(it))) + return patch( + "OWNd.connection.event_session.time", + SimpleNamespace(monotonic=lambda: next(it)), + ) def _dropping_event_session() -> tuple[OWNEventSession, MagicMock]: @@ -629,3 +632,181 @@ async def test_command_session_rejects_invalid_or_empty_message() -> None: assert await session.send("*#123456789##") is None assert logger.error.call_count == 6 + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "frame", + [ + "*1*1*11##", + "*#4*000*0*0215##", + "*#4*000*#0*0215##", + ], +) +async def test_written_non_query_with_status_request_flag_is_not_replayed_after_lost_ack( + frame: str, +) -> None: + session, writer = make_session(OWNCommandSession) + assert isinstance(session, OWNCommandSession) + assert session._stream_reader is not None + session._stream_reader.feed_eof() + + with patch.object(session, "connect", new_callable=AsyncMock) as connect: + result = await session.send(frame, is_status_request=True) + + assert result is None + assert writer.written == [frame.encode()] + connect.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_disconnected_dimension_request_retries_once() -> None: + session, original_writer = make_session(OWNCommandSession) + assert isinstance(session, OWNCommandSession) + assert session._stream_reader is not None + session._stream_reader.feed_eof() + + retry_writer = FakeWriter() + + async def reconnect() -> dict[str, object]: + session._stream_reader = asyncio.StreamReader() + session._stream_reader.feed_eof() + session._stream_writer = retry_writer # type: ignore[assignment] + return {"Success": True, "Message": None} + + with patch.object(session, "connect", side_effect=reconnect) as connect: + result = await session.send("*#4*000*0##", is_status_request=True) + + assert result is None + connect.assert_awaited_once() + assert original_writer.written == [b"*#4*000*0##"] + assert retry_writer.written == [b"*#4*000*0##"] + + +@pytest.mark.asyncio +async def test_typed_exceptions_and_last_error() -> None: + from OWNd import ( + OWNAuthError, + OWNConnectionError, + OWNError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, + ) + + session, _ = make_session(OWNCommandSession) + assert isinstance(session, OWNCommandSession) + + # 1. Invalid frame -> OWNProtocolError + with pytest.raises(OWNProtocolError) as exc_info: + await session.send("not-a-frame", raise_on_error=True) + assert isinstance(session.last_error, OWNProtocolError) + assert exc_info.value.frame == "not-a-frame" + + # 2. NACK response -> OWNNackError + assert session._stream_reader is not None + session._stream_reader.feed_data(b"*#*0##") + with pytest.raises(OWNNackError) as nack_exc: + await session.send("*#1*11##", is_status_request=True, raise_on_error=True) + assert isinstance(session.last_error, OWNNackError) + assert nack_exc.value.frame == "*#1*11##" + + # 3. Timeout response -> OWNTimeoutError (subclass of OWNConnectionError and OWNError) + session2, _ = make_session(OWNCommandSession) + assert isinstance(session2, OWNCommandSession) + session2._read_command_response = AsyncMock(side_effect=TimeoutError("Timed out")) # type: ignore[method-assign] + with pytest.raises(OWNTimeoutError): + await session2.send("*1*1*11##", raise_on_error=True) + assert isinstance(session2.last_error, OWNTimeoutError) + assert isinstance(session2.last_error, OWNConnectionError) + assert isinstance(session2.last_error, OWNError) + + # 4. Auth failure on connect(raise_on_error=True) -> OWNAuthError + auth_seen: list[OWNAuthError] = [] + session3, _ = make_session(OWNCommandSession) + session3.on_auth_error = auth_seen.append + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object( + session3, + "_negotiate", + new=AsyncMock( + return_value={"Success": False, "Message": "password_error"} + ), + ), + ): + with pytest.raises(OWNAuthError): + await session3.connect(raise_on_error=True) + assert len(auth_seen) == 1 + assert auth_seen[0].reason == "password_error" + assert isinstance(session3.last_error, OWNAuthError) + + +@pytest.mark.asyncio +async def test_event_session_raise_on_auth_error() -> None: + from OWNd import OWNAuthError + + auth_seen: list[OWNAuthError] = [] + session, _ = make_session(OWNEventSession) + assert isinstance(session, OWNEventSession) + session.on_auth_error = auth_seen.append + session._stream_reader = None + session._stream_writer = None + + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object( + session, + "_negotiate", + new=AsyncMock( + return_value={"Success": False, "Message": "password_required"} + ), + ), + ): + with pytest.raises(OWNAuthError) as exc_info: + await session.get_next(raise_on_auth_error=True) + + assert exc_info.value.reason == "password_required" + assert len(auth_seen) == 1 + + +@pytest.mark.asyncio +async def test_send_when_disconnected_uses_single_connect_attempt_and_cooldown() -> None: + session, _ = make_session(OWNCommandSession) + assert isinstance(session, OWNCommandSession) + session._stream_reader = None + session._stream_writer = None + + open_conn = AsyncMock(side_effect=ConnectionRefusedError) + with patch("OWNd.connection.session.asyncio.open_connection", new=open_conn): + # First send() attempts only 1 TCP connect (not 5 with backoff) + assert await session.send("*1*1*11##") is None + assert open_conn.await_count == 1 + + # Second send() within cooldown window fails fast without calling open_connection again + assert await session.send("*1*1*12##") is None + assert open_conn.await_count == 1 + + +def test_gateway_clock_builders_accept_tzinfo() -> None: + import datetime as dt + from OWNd.message import OWNGatewayCommand + + tz = dt.timezone(dt.timedelta(hours=2)) + cmd_dt = OWNGatewayCommand.set_datetime_to_now(tz) + cmd_d = OWNGatewayCommand.set_date_to_today(tz) + cmd_t = OWNGatewayCommand.set_time_to_now(tz) + assert str(cmd_dt).startswith("*#13**#22*") + assert "*002*" in str(cmd_dt) + assert str(cmd_d).startswith("*#13**#1*0") + assert str(cmd_t).startswith("*#13**#0*") + assert str(cmd_t).endswith("*002##") + with pytest.raises(ValueError, match="time_zone must be a datetime.tzinfo"): + OWNGatewayCommand.set_datetime_to_now(123) # type: ignore[arg-type] + + diff --git a/tests/test_connection_extended.py b/tests/test_connection_extended.py index 59306bd4..a7366cce 100644 --- a/tests/test_connection_extended.py +++ b/tests/test_connection_extended.py @@ -667,14 +667,14 @@ class TestOWNGatewayDiscoveryAndClassmethods: @pytest.mark.asyncio async def test_get_first_available_gateway(self): - with patch("OWNd.connection.find_gateways", return_value=[{"address": "192.168.1.50"}]): + with patch("OWNd.connection.gateway.find_gateways", return_value=[{"address": "192.168.1.50"}]): gw = await OWNGateway.get_first_available_gateway(password="secret") assert gw.host == "192.168.1.50" assert gw.password == "secret" @pytest.mark.asyncio async def test_find_from_address_present(self): - with patch("OWNd.connection.get_gateway", return_value={"address": "192.168.1.55"}): + with patch("OWNd.connection.gateway.get_gateway", return_value={"address": "192.168.1.55"}): gw = await OWNGateway.find_from_address("192.168.1.55") assert gw.host == "192.168.1.55" @@ -701,7 +701,7 @@ async def test_build_from_discovery_info_fetch_port(self): "ssdp_location": "http://192.168.1.100:8080/desc.xml", "port": None, } - with patch("OWNd.connection.get_port", return_value=20000): + with patch("OWNd.connection.gateway.get_port", return_value=20000): gw = await OWNGateway.build_from_discovery_info(info) assert gw.port == 20000 @@ -1279,23 +1279,3 @@ async def fake_keepalive(): assert call_count >= 2 - -def test_connection_module_delattr(): - """Verify that deleting an attribute on OWNd.connection propagates to submodules.""" - import OWNd.connection as conn - import OWNd.connection.gateway as gw_mod - - conn._custom_transient_attr = "hello" - assert hasattr(conn, "_custom_transient_attr") - del conn._custom_transient_attr - assert not hasattr(conn, "_custom_transient_attr") - - # Also test deleting an attribute present on a submodule - gw_mod._dummy_test_attr = 123 - conn._dummy_test_attr = 456 - del conn._dummy_test_attr - assert not hasattr(gw_mod, "_dummy_test_attr") - - - - diff --git a/tests/test_connection_faults.py b/tests/test_connection_faults.py index 0dbd23d3..4a82d777 100644 --- a/tests/test_connection_faults.py +++ b/tests/test_connection_faults.py @@ -74,7 +74,7 @@ class TestConnectionFaults: @pytest.mark.asyncio async def test_get_first_available_gateway_empty(self) -> None: - with patch("OWNd.connection.find_gateways", new=AsyncMock(return_value=[])): + with patch("OWNd.connection.gateway.find_gateways", new=AsyncMock(return_value=[])): result = await OWNGateway.get_first_available_gateway() assert result is None @@ -155,7 +155,7 @@ async def test_connect_network_oserror_retry(self) -> None: with ( patch("asyncio.open_connection", side_effect=OSError("Network unreachable")), patch("asyncio.sleep", new=AsyncMock()), - patch("OWNd.connection.MAX_CONNECT_ATTEMPTS", 1), + patch("OWNd.connection.session.MAX_CONNECT_ATTEMPTS", 1), ): result = await session.connect() assert result is None @@ -166,7 +166,7 @@ async def test_negotiation_max_frames_exhaustion(self) -> None: session, writer = make_fault_session(password="12345") # Gateway sends ACK then SHA challenge, but with max_frames=1, second read exceeds limit session._read_frame = AsyncMock(side_effect=["*#*1##", "*98*1##"]) - with patch("OWNd.connection.NEGOTIATION_MAX_FRAMES", 1): + with patch("OWNd.connection.session.NEGOTIATION_MAX_FRAMES", 1): result = await session._negotiate() assert result == {"Success": False, "Message": "negotiation_timeout"} @@ -272,7 +272,7 @@ async def test_command_session_read_response_branches(self) -> None: assert sig2.is_ack() # 3. Malformed frame raises ValueError during OWNMessage.parse (lines 1217-1224) - with patch("OWNd.connection.OWNMessage.parse", side_effect=[ValueError("Malformed"), OWNSignaling("*#*1##")]): + with patch("OWNd.connection.command_session.OWNMessage.parse", side_effect=[ValueError("Malformed"), OWNSignaling("*#*1##")]): session._read_frame = AsyncMock(side_effect=["*#bad##", "*#*1##"]) sig3, collected3 = await session._read_command_response() assert sig3.is_ack() @@ -316,7 +316,7 @@ async def test_connect_non_fatal_negotiation_failure(self) -> None: patch.object(session, "_negotiate", new=AsyncMock(return_value={"Success": False, "Reason": "temporary_busy", "Message": "Busy"})), patch("asyncio.open_connection", new=AsyncMock(return_value=(MagicMock(), writer))), patch("asyncio.sleep", new=AsyncMock()), - patch("OWNd.connection.MAX_CONNECT_ATTEMPTS", 2), + patch("OWNd.connection.session.MAX_CONNECT_ATTEMPTS", 2), ): res = await session.connect() assert res is None diff --git a/tests/test_connection_proxy.py b/tests/test_connection_proxy.py deleted file mode 100644 index 5b3f61ee..00000000 --- a/tests/test_connection_proxy.py +++ /dev/null @@ -1,153 +0,0 @@ -"""Tests verifying OWNd.connection proxy module behavior, attribute forwarding, and reload.""" -from __future__ import annotations - -import importlib -from types import ModuleType -from unittest.mock import MagicMock, patch -import pytest - -import OWNd.connection as conn -from OWNd.connection import _FORWARD_TARGETS, _ConnectionModule - - -def test_connection_proxy_dynamic_submodules() -> None: - """Verify _get_submodules dynamically resolves loaded connection submodules.""" - assert isinstance(conn, _ConnectionModule) - submodules = conn._get_submodules() - assert len(submodules) >= 5 - mod_names = [m.__name__.split(".")[-1] for m in submodules] - for expected in ("auth", "command_session", "event_session", "gateway", "session"): - assert expected in mod_names - - -def test_connection_proxy_getattr_from_submodule() -> None: - """Verify __getattr__ dynamically retrieves attributes from submodules.""" - import OWNd.connection.session as sess_mod - - sess_mod._test_dynamic_probe_value = "probe_found" - try: - assert getattr(conn, "_test_dynamic_probe_value") == "probe_found" - finally: - del sess_mod._test_dynamic_probe_value - - -def test_connection_proxy_getattr_nonexistent() -> None: - """Verify __getattr__ raises AttributeError for non-existent attributes.""" - with pytest.raises(AttributeError, match="has no attribute 'nonexistent_test_attr_xyz'"): - _ = conn.nonexistent_test_attr_xyz - - -def test_connection_proxy_dir() -> None: - """Verify __dir__ lists attributes from proxy and underlying submodules.""" - dir_list = dir(conn) - assert "OWNSession" in dir_list - assert "calculate_open_password" in dir_list - assert "find_gateways" in dir_list - - -def test_connection_proxy_setattr_forwarding() -> None: - """Verify setting patched attributes propagates to all submodules possessing them.""" - import OWNd.connection.gateway as gw_mod - - orig = gw_mod.find_gateways - dummy_finder = MagicMock(return_value=["mocked"]) - try: - conn.find_gateways = dummy_finder - assert gw_mod.find_gateways is dummy_finder - assert conn.find_gateways is dummy_finder - finally: - conn.find_gateways = orig - assert gw_mod.find_gateways is orig - - -def test_connection_proxy_setattr_and_delattr_dunder() -> None: - """Verify dunder attributes are set and deleted on proxy without submodule forwarding.""" - conn.__test_dunder_proxy__ = "dunder_val" - assert conn.__test_dunder_proxy__ == "dunder_val" - del conn.__test_dunder_proxy__ - assert not hasattr(conn, "__test_dunder_proxy__") - - -def test_connection_proxy_delattr_submodule() -> None: - """Verify deleting an attribute removes it from proxy and submodules.""" - import OWNd.connection.gateway as gw_mod - - gw_mod._transient_submodule_attr = "sub_val" - conn._transient_submodule_attr = "sub_val" - assert hasattr(gw_mod, "_transient_submodule_attr") - - del conn._transient_submodule_attr - assert not hasattr(conn, "_transient_submodule_attr") - assert not hasattr(gw_mod, "_transient_submodule_attr") - - -def test_connection_proxy_delattr_submodule_only() -> None: - """Verify deleting an attribute that exists only on a submodule (not on proxy dict) succeeds.""" - import OWNd.connection.gateway as gw_mod - - gw_mod._submodule_only_probe = "probe_val" - try: - assert hasattr(conn, "_submodule_only_probe") - assert "_submodule_only_probe" not in conn.__dict__ - del conn._submodule_only_probe - assert not hasattr(conn, "_submodule_only_probe") - assert not hasattr(gw_mod, "_submodule_only_probe") - finally: - if hasattr(gw_mod, "_submodule_only_probe"): - del gw_mod._submodule_only_probe - - -def test_connection_proxy_delattr_nonexistent() -> None: - """Verify deleting a nonexistent attribute raises AttributeError.""" - with pytest.raises(AttributeError, match="has no attribute 'nonexistent_attr_xyz'"): - del conn.nonexistent_attr_xyz - - -def test_connection_proxy_forward_targets_missing_fails_loudly() -> None: - """Verify that if a required forward target is missing from the submodule, it fails loudly.""" - _FORWARD_TARGETS["test_missing_forward_target"] = ("session",) - try: - with pytest.raises( - AttributeError, - match="Patched attribute 'test_missing_forward_target' missing from expected target submodule 'session'", - ): - conn.test_missing_forward_target = "should_fail" - assert "test_missing_forward_target" not in conn.__dict__ - finally: - _FORWARD_TARGETS.pop("test_missing_forward_target", None) - - -def test_connection_proxy_forward_targets_submodule_missing_fails_loudly() -> None: - """Verify that if the target submodule itself is not loaded/missing, it fails loudly.""" - _FORWARD_TARGETS["test_missing_submodule"] = ("nonexistent_submodule",) - try: - with pytest.raises( - AttributeError, - match="Patched attribute 'test_missing_submodule' missing from expected target submodule 'nonexistent_submodule'", - ): - conn.test_missing_submodule = "should_fail" - assert "test_missing_submodule" not in conn.__dict__ - finally: - _FORWARD_TARGETS.pop("test_missing_submodule", None) - - -def test_connection_proxy_all_attribute_missing_fails_loudly() -> None: - """Verify that setting an attribute declared in __all__ that is missing from all submodules fails loudly.""" - all_list = list(conn.__all__) - all_list.append("bogus_all_symbol_missing") - with patch.object(conn, "__all__", all_list): - with pytest.raises( - AttributeError, - match="Patched attribute 'bogus_all_symbol_missing' in __all__ is missing from all connection submodules", - ): - conn.bogus_all_symbol_missing = "fail" - assert "bogus_all_symbol_missing" not in conn.__dict__ - - -def test_connection_proxy_reload() -> None: - """Verify that importlib.reload reloads cleanly, preserves _ConnectionModule, and restores state.""" - reloaded = importlib.reload(conn) - assert isinstance(reloaded, reloaded._ConnectionModule) - assert reloaded.OWNSession is not None - assert reloaded.find_gateways is not None - diff --git a/tests/test_message_exhaustive.py b/tests/test_message_exhaustive.py index 388ba080..734bb314 100644 --- a/tests/test_message_exhaustive.py +++ b/tests/test_message_exhaustive.py @@ -574,7 +574,11 @@ def test_message_remaining_branches(self) -> None: # Gateway command dimension 22 gw_cmd22 = OWNGatewayCommand("*#13**22*12*30*00*001*0*10*09*2026##") assert gw_cmd22._datetime is not None + assert str(OWNGatewayCommand.set_datetime_to_now(datetime.timezone.utc)).startswith("*#13**") + assert str(OWNGatewayCommand.set_date_to_today(datetime.timezone.utc)).startswith("*#13**") + assert str(OWNGatewayCommand.set_time_to_now(datetime.timezone.utc)).startswith("*#13**") # Sound command status assert str(OWNSoundCommand.status("21")) == "*#16*21*5##" + diff --git a/tests/typing/consumer_send.py b/tests/typing/consumer_send.py index 941d5be3..3432c0a1 100644 --- a/tests/typing/consumer_send.py +++ b/tests/typing/consumer_send.py @@ -2,10 +2,11 @@ OWNd ships ``py.typed``, so its annotations are the contract MyHOME and every other consumer type-check against. This module is the consumer's -side of that contract for the command session: each function below is a -call MyHOME really makes, and ``mypy --strict`` over this file (see the -``quality`` job in CI, and ``tests/test_typing_consumer.py``) fails the -build if a signature change would reject it. +side of that contract for the command session and transport abstractions: +each function below is a call MyHOME really makes, and ``mypy --strict`` +over this file (see the ``quality`` job in CI, and +``tests/test_typing_consumer.py``) fails the build if a signature change +would reject it. Keep it to public API and to calls that must keep working; a private name here would pin an implementation detail. @@ -19,6 +20,7 @@ OWNMessage, OWNSceneCommand, ) +from OWNd.transport import AsyncSerialTransport, AsyncTcpTransport, OWNTransport async def send_raw_frame(session: OWNCommandSession) -> None: @@ -52,3 +54,19 @@ async def one_shot(gateway: OWNGateway) -> None: """The connect-send-close helper accepts the same two forms.""" await OWNCommandSession.send_to_gateway("*1*0*11##", gateway) await OWNCommandSession.send_to_gateway(OWNLightingCommand.switch_off("11"), gateway) + + +async def send_via_transport( + transport: OWNTransport, + tcp_transport: AsyncTcpTransport, + serial_transport: AsyncSerialTransport, + frame: str, +) -> None: + """Transport implementations accept both raw frame strings and OWNMessage instances.""" + await transport.send("*1*1*11##") + await transport.send(OWNLightingCommand.switch_on("11")) + await tcp_transport.send(OWNLightingCommand.status("11"), is_status_request=True) + await serial_transport.send(OWNSceneCommand.start("1")) + parsed = OWNMessage.parse(frame) + if parsed is not None: + await transport.send(parsed) diff --git a/tools/golden/MASSI_OUTREACH.md b/tools/golden/MASSI_OUTREACH.md deleted file mode 100644 index 0224fc0b..00000000 --- a/tools/golden/MASSI_OUTREACH.md +++ /dev/null @@ -1,54 +0,0 @@ -# Massi Valla (mvalla) Outreach Template - -> **Context**: Draft message to reach out to Massi Valla (author/maintainer of `openwebnet4j` and openHAB OpenWebNet binding). -> **Status**: On hold as requested while corpus harvesting and internal alignment expand. -> **Objective**: Propose an open, cross-community "OpenWebNet Golden Corpus" to align frame parsing/building across openHAB, Home Assistant, and Legrand specifications. - ---- - -### Subject: Cross-ecosystem OpenWebNet Golden Corpus (openHAB & Home Assistant conformance) - -Hi Massi, - -First off, huge thank you for your decade of incredible work on `openwebnet4j` and the openHAB OpenWebNet binding. Your library has long been the gold standard for reliable OpenWebNet communication in open-source home automation. - -Over in the Home Assistant MyHOME integration (`OpenWebNet-HA/MyHOME`), we've been building a standardized **OpenWebNet Golden Corpus**—a declarative, language-agnostic test suite of frames in YAML/JSON. - -The core motivation is simple: OpenWebNet frame edge cases (private bus `#4#bus` routing, leading zeros in A/PL addresses like `0311`, dimension writes, and CEN/CEN+ presses) often lead to subtle divergence across Python, Java, and other implementations. - -### How it works -We use a 3-authority conformance model: -1. **Judge**: Official Legrand specification PDFs (verified via the `openwebnet-mcp` protocol engine). -2. **Oracle**: `openwebnet4j` & openHAB reference vectors (`MessageTest.java` and `OwnIdTest.java`). -3. **SUT**: Our Python runtime (`OWNd` / `MyHOME`). - -We have harvested and cross-validated **58 fixtures across 11 OpenWebNet subsystems** (Lighting, Automation, Thermoregulation, Alarm, Auxiliary, Gateway Mgmt, CEN, CEN+, Energy, Scenarios, Signaling) with a 100% roundtrip pass rate (137 automated assertions). - -Each fixture is validated against a strict JSON Schema: - -```yaml -- id: light.cmd.on.bus.0311 - frame: "*1*1*0311#4#01##" - direction: command - who: 1 - what: 1 - where: "0311" - interface: "01" - source: openwebnet4j - mcp_valid: true - mcp_notes: "Lighting: Turn ON at address '0311' routed to private SCS bus 01" - roundtrip: true -``` - -### Proposal & Collaboration -We would love your thoughts on making this golden corpus an independent, shared standard that benefits both openHAB and Home Assistant: -1. **Schema Review**: Would you be open to reviewing the schema structure to ensure it accommodates all nuances of `openwebnet4j`? -2. **Shared Conformance**: If you think this is worthwhile, we would be delighted to do the legwork to contribute a test runner or exporter to `openwebnet4j` so both projects can run against the exact same declarative conformance suite. - -Zero obligation or pressure, of course. If you're interested, you can explore the schema, harvest report, and fixtures here: -https://github.com/OpenWebNet-HA/MyHOME/tree/v2-phase1-architecture/tests/golden - -Thanks again for all your pioneering work for the MyHOME community! - -Best regards, -The OpenWebNet-HA Team diff --git a/tools/golden/MCP_NOTES.md b/tools/golden/MCP_NOTES.md deleted file mode 100644 index 5bd9eae7..00000000 --- a/tools/golden/MCP_NOTES.md +++ /dev/null @@ -1,60 +0,0 @@ -# OpenWebNet MCP Server Notes & Verification Log - -- **Server**: `openwebnet-mcp` -- **Repository**: `https://github.com/OpenWebNet-HA/openwebnet-mcp.git` -- **Date Verified**: 2026-09-11 -- **Client**: Antigravity Pair-Programming Agent on `OpenWebNet-HA/MyHOME` (`v2-phase1-architecture`) - ---- - -## 1. Tool Catalog & Availability - -| Tool | Status | Purpose | Verified Input / Test Case | -|---|---|---|---| -| `list_who_catalog` | **PASS** | Returns master table of all 21 OpenWebNet WHO subsystems with Legrand titles, Home Assistant entities, and status | Invoked with `{}` | -| `get_who_spec` | **PASS** | Retrieves full specification, addressing rules, WHAT codes, and dimensions for a given WHO | Verified with `who: 1` (Lighting v1.1.0) and `who: 2` (Automation v1.0.0) | -| `parse_and_validate_frame` | **PASS** | Parses raw OWN frame, extracts semantic components, validates against official Legrand specifications, emits warnings for non-standard frames | Verified with `*1*1*51##`, `*#1*51##`, `*1*1*0311#4#01##`, `*2*1*93##`, `*#*1##`, `*#*0##`, and invalid syntax `*1*1*51` | -| `draft_own_frame` | **PASS** | Constructs and validates syntactically correct OpenWebNet frames from parameters | Verified for light ON (`*1*1*51##`) and cover STOP (`*2*0*93##`) | -| `lookup_frame_syntax` | **PASS** | Provides grammar regex patterns, templates, and parameter formats for all frame types | Verified for `who: 1` | -| `search_documentation` | **PASS** | Searches indexed Legrand documentation PDFs | Available | -| `get_ha_guide` | **PASS** | HA integration guidelines for OpenWebNet subsystems | Available | -| `get_code_signature` | **PASS** | Extracts code signatures from reference implementations | Available | -| `rescan_documentation` | **PASS** | Reloads documentation cache | Available | - ---- - -## 2. Detailed Behavior Notes & Verification Log - -### `list_who_catalog` -- Returned cleanly formatted markdown table with 21 WHO families (0, 1, 2, 3, 4, 5, 6, 7, 9, 13, 14, 15, 16, 17, 18, 22, 24, 25, 1001, 1004, 1013). -- Zero hallucination; matches BTicino specification categories. - -### `get_who_spec` -- **WHO=1 (Lighting)**: - - Spec: `WHO_1.pdf` v1.1.0 (2014-11-17). - - Documents WHAT codes: `0` (OFF), `1` (ON 100%), `2`-`10` (stepped dimming 20%-100%), `11`/`12` (dim UP/DOWN), `13` (toggle), `14` (timed), `15` (blinking), `1#` (ON with transition speed). - - Documents Dimensions: Dimension 1 (level 1-100%), Dimension 2 (RGB), Dimension 3 (Tunable White Kelvin). -- **WHO=2 (Automation / Covers)**: - - Spec: `WHO_2.pdf` v1.0.0 (2015-11-12). - - Documents WHAT codes: `0` (STOP), `1` (UP), `2` (DOWN), `30` (Adv STOP), `31` (Adv UP), `32` (Adv DOWN). - - Documents Dimensions: Dimension 10 (Position percentage 0-100%), Dimension 11 (Louvre tilt angle 0-100%). - -### `parse_and_validate_frame` -- Point-to-Point Light ON: `*1*1*51##` -> Valid: YES, Type: `STATUS_EVENT`, Subsystem: `1`, Address: `51`, Action: `1`. -- Status Request: `*#1*51##` -> Valid: YES, Type: `STATUS_REQUEST`, Subsystem: `1`, Address: `51`. -- Local Bus Routed Light: `*1*1*0311#4#01##` -> Valid: YES, Type: `STATUS_EVENT`, Subsystem: `1`, Address: `0311`, Action: `1`, routed to private SCS bus 01. -- Automation UP: `*2*1*93##` -> Valid: YES, Type: `STATUS_EVENT`, Subsystem: `2`, Address: `93`, Action: `1`. -- Signaling ACK: `*#*1##` -> Valid: YES, Type: `ACK`, Subsystem: `None`, Address: `N/A`. -- Signaling NACK: `*#*0##` -> Valid: YES, Type: `NACK`, Subsystem: `None`, Address: `N/A`. -- Non-standard WHAT: `*1*999*51##` -> Valid: YES, Warnings: `WHAT=999 is not standard for WHO=1`. -- Malformed Delimiters: `*1*1*51` -> Valid: NO, Type: `UNKNOWN`, Warnings: `Frame must begin with '*' and end with '##'`. - -### `draft_own_frame` -- `{who: 1, command_type: "command", what: 1, where: "51"}` -> `*1*1*51##` (Valid: YES) -- `{who: 2, command_type: "command", what: 0, where: "93"}` -> `*2*0*93##` (Valid: YES) - ---- - -## 3. Exit Criteria Evaluation -- Agent can parse and validate any OpenWebNet frame against official Legrand specifications via `openwebnet-mcp`. -- All tools function deterministically and quickly. From 6fbd6fd9021524681e49f72f40de38685b9cc193 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:47:01 +0200 Subject: [PATCH 2/2] test(connection): achieve 100% statement coverage on typed exception paths and preserve discovery patch compatibility --- OWNd/connection/command_session.py | 12 +-- OWNd/connection/event_session.py | 53 +++++----- OWNd/connection/gateway.py | 20 +++- OWNd/connection/session.py | 4 +- tests/test_connection.py | 163 +++++++++++++++++++++++++++++ 5 files changed, 211 insertions(+), 41 deletions(-) diff --git a/OWNd/connection/command_session.py b/OWNd/connection/command_session.py index ce793cef..dfcfb173 100644 --- a/OWNd/connection/command_session.py +++ b/OWNd/connection/command_session.py @@ -57,12 +57,8 @@ def __init__( async def connect( self, *, raise_on_error: bool = False ) -> dict[str, Any] | None: - if self._connect_max_attempts_override is None: - self._last_reconnect_failure = None - result = await super().connect(raise_on_error=raise_on_error) - if result is not None and result.get("Success", False): - self._last_reconnect_failure = None - return result + self._last_reconnect_failure = None + return await super().connect(raise_on_error=raise_on_error) @classmethod async def send_to_gateway( @@ -89,9 +85,7 @@ async def connect_to_gateway( ) -> dict[str, Any] | None: connection = cls(gateway) try: - if raise_on_error: - return await connection.connect(raise_on_error=True) - return await connection.connect() + return await connection.connect(raise_on_error=raise_on_error) finally: with contextlib.suppress(Exception): await connection.close() diff --git a/OWNd/connection/event_session.py b/OWNd/connection/event_session.py index 772a72fb..527649dd 100644 --- a/OWNd/connection/event_session.py +++ b/OWNd/connection/event_session.py @@ -111,9 +111,7 @@ async def connect_to_gateway( ) -> dict[str, Any] | None: connection = cls(gateway) try: - if raise_on_error: - return await connection.connect(raise_on_error=True) - return await connection.connect() + return await connection.connect(raise_on_error=raise_on_error) finally: with contextlib.suppress(Exception): await connection.close() @@ -130,19 +128,31 @@ def _maybe_raise_auth_error( ) if not should_raise: return - if isinstance(self._last_error, OWNAuthError): - raise self._last_error - if ( - result is not None - and result.get("Message") in _AUTH_NEGOTIATION_ERRORS - ): - reason = str(result.get("Message")) - err = OWNAuthError( - f"Gateway authentication failed ({reason})", reason=reason + err = ( + self._last_error + if isinstance(self._last_error, OWNAuthError) + else ( + OWNAuthError( + f"Gateway authentication failed ({result.get('Message')})", + reason=str(result.get("Message")), + ) + if result is not None + and result.get("Message") in _AUTH_NEGOTIATION_ERRORS + else None ) + ) + if err is not None: self._last_error = err raise err + async def _reconnect_and_check_auth( + self, raise_on_auth_error: bool | None + ) -> dict[str, Any] | None: + result = await self._reconnect() + if self._stream_reader is None: + self._maybe_raise_auth_error(result, raise_on_auth_error) + return result + async def get_next( self, *, raise_on_auth_error: bool | None = None ) -> OWNMessage | str | None: @@ -163,9 +173,8 @@ async def get_next( "%s Event session not connected, reconnecting...", self._log_id, ) - result = await self._reconnect() + result = await self._reconnect_and_check_auth(raise_on_auth_error) if self._stream_reader is None: - self._maybe_raise_auth_error(result, raise_on_auth_error) # Still no stream: connect() gave up or failed fatally and # returned IMMEDIATELY (its internal back-off only covers the # transient retries). Pause here, otherwise the caller's read @@ -196,18 +205,14 @@ async def get_next( self._log_id, self._inactivity_timeout, ) - result = await self._reconnect() - if self._stream_reader is None: - self._maybe_raise_auth_error(result, raise_on_auth_error) + await self._reconnect_and_check_auth(raise_on_auth_error) return None except asyncio.LimitOverrunError: self._logger.warning( "%s Received oversized or garbage frame, dropping connection and reconnecting...", self._log_id, ) - result = await self._reconnect() - if self._stream_reader is None: - self._maybe_raise_auth_error(result, raise_on_auth_error) + await self._reconnect_and_check_auth(raise_on_auth_error) return None except ( asyncio.IncompleteReadError, @@ -243,17 +248,13 @@ async def get_next( self._logger.debug( "%s Event connection lost, reconnecting...", self._log_id ) - result = await self._reconnect() - if self._stream_reader is None: - self._maybe_raise_auth_error(result, raise_on_auth_error) + await self._reconnect_and_check_auth(raise_on_auth_error) return None except Exception: # pylint: disable=broad-except self._logger.exception( "%s Event session crashed, reconnecting...", self._log_id ) - result = await self._reconnect() - if self._stream_reader is None: - self._maybe_raise_auth_error(result, raise_on_auth_error) + await self._reconnect_and_check_auth(raise_on_auth_error) return None # A frame was read successfully: from here on, any failure is a diff --git a/OWNd/connection/gateway.py b/OWNd/connection/gateway.py index 996c6842..14e0355f 100644 --- a/OWNd/connection/gateway.py +++ b/OWNd/connection/gateway.py @@ -4,12 +4,23 @@ from collections.abc import Mapping import logging +import sys from typing import Any from urllib.parse import urlparse from ..discovery import find_gateways, get_gateway, get_port from ..profiles import GatewayProfile, get_gateway_profile +_ORIG_FIND_GATEWAYS = find_gateways +_ORIG_GET_GATEWAY = get_gateway +_ORIG_GET_PORT = get_port + + +def _resolve_discovery_fn(name: str, local_fn: Any, orig_fn: Any) -> Any: + """Resolve discovery helper, honoring overrides on either gateway or connection.""" + pkg = sys.modules.get(__package__ or "") + return getattr(pkg, name, local_fn) if local_fn is orig_fn else local_fn + def _first_scalar(value: Any, default: Any = None) -> Any: """Return a scalar from legacy tuple/list discovery values.""" @@ -117,7 +128,8 @@ def log_id(self, value: str) -> None: async def get_first_available_gateway( cls, password: str | None = None ) -> OWNGateway | None: - local_gateways = await find_gateways() + finder = _resolve_discovery_fn("find_gateways", find_gateways, _ORIG_FIND_GATEWAYS) + local_gateways = await finder() if not local_gateways: return None local_gateways[0]["password"] = password @@ -126,7 +138,8 @@ async def get_first_available_gateway( @classmethod async def find_from_address(cls, address: str | None) -> OWNGateway | None: if address is not None: - gateway = await get_gateway(address) + getter = _resolve_discovery_fn("get_gateway", get_gateway, _ORIG_GET_GATEWAY) + gateway = await getter(address) return cls(gateway) if gateway is not None else None return await cls.get_first_available_gateway() @@ -150,7 +163,8 @@ async def build_from_discovery_info( "ssdp_location" in discovery_info and discovery_info["ssdp_location"] is not None ): - discovery_info["port"] = await get_port(discovery_info["ssdp_location"]) + port_getter = _resolve_discovery_fn("get_port", get_port, _ORIG_GET_PORT) + discovery_info["port"] = await port_getter(discovery_info["ssdp_location"]) elif "address" in discovery_info and discovery_info["address"] is not None: return await cls.find_from_address(discovery_info["address"]) else: diff --git a/OWNd/connection/session.py b/OWNd/connection/session.py index 952fec28..512a25eb 100644 --- a/OWNd/connection/session.py +++ b/OWNd/connection/session.py @@ -317,9 +317,7 @@ async def test_gateway( cls, gateway: OWNGateway, *, raise_on_error: bool = False ) -> dict[str, Any]: connection = cls(gateway) - if raise_on_error: - return await connection.test_connection(raise_on_error=True) - return await connection.test_connection() + return await connection.test_connection(raise_on_error=raise_on_error) async def test_connection( self, *, raise_on_error: bool = False diff --git a/tests/test_connection.py b/tests/test_connection.py index ad77eae3..faf16141 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -810,3 +810,166 @@ def test_gateway_clock_builders_accept_tzinfo() -> None: OWNGatewayCommand.set_datetime_to_now(123) # type: ignore[arg-type] +@pytest.mark.asyncio +async def test_session_and_command_error_paths_with_raise_on_error() -> None: + from OWNd import ( + OWNAuthError, + OWNConnectionError, + OWNNackError, + OWNProtocolError, + OWNTimeoutError, + ) + from OWNd.connection.session import _negotiation_error_to_exception + from OWNd.message import OWNSignaling + + # _negotiation_error_to_exception branches + assert isinstance(_negotiation_error_to_exception("negotiation_timeout"), OWNTimeoutError) + assert isinstance(_negotiation_error_to_exception("negotiation_refused"), OWNNackError) + assert isinstance(_negotiation_error_to_exception("negotiation_error"), OWNProtocolError) + assert isinstance(_negotiation_error_to_exception(None), OWNConnectionError) + + # on_auth_error getter & exception resilience in _notify_auth_error + session, _ = make_session(OWNCommandSession) + assert session.on_auth_error is None + + def broken_cb(err: OWNAuthError) -> None: + raise RuntimeError("callback boom") + + session.on_auth_error = broken_cb + assert session.on_auth_error is broken_cb + session._notify_auth_error(OWNAuthError("bad pw", reason="password_error")) + + # test_connection(raise_on_error=True) refused after 3 attempts + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(side_effect=ConnectionRefusedError), + ), + patch("OWNd.connection.session.asyncio.sleep", new=AsyncMock()), + ): + with pytest.raises(OWNConnectionError): + await session.test_connection(raise_on_error=True) + + # test_connection(raise_on_error=True) negotiation failure (auth & non-auth) + auth_called: list[OWNAuthError] = [] + session.on_auth_error = auth_called.append + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object( + session, + "_negotiate", + new=AsyncMock(return_value={"Success": False, "Message": "password_error"}), + ), + ): + with pytest.raises(OWNAuthError): + await session.test_connection(raise_on_error=True) + assert len(auth_called) == 1 + + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object( + session, + "_negotiate", + new=AsyncMock(return_value={"Success": False, "Message": "negotiation_refused"}), + ), + ): + with pytest.raises(OWNNackError): + await session.test_connection(raise_on_error=True) + + # test_connection(raise_on_error=True) ConnectionResetError and TimeoutError during _negotiate + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object(session, "_negotiate", new=AsyncMock(side_effect=ConnectionResetError)), + ): + with pytest.raises(OWNConnectionError) as rst_exc: + await session.test_connection(raise_on_error=True) + assert rst_exc.value.reason == "password_retry" + + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object(session, "_negotiate", new=AsyncMock(side_effect=TimeoutError("slow"))), + ): + with pytest.raises(OWNTimeoutError): + await session.test_connection(raise_on_error=True) + + # connect(raise_on_error=True) exhausting max_attempts, and send() reconnect + cooldown raising + cmd_sess, _ = make_session(OWNCommandSession) + assert isinstance(cmd_sess, OWNCommandSession) + cmd_sess._stream_reader = None + cmd_sess._stream_writer = None + with patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(side_effect=ConnectionRefusedError), + ): + cmd_sess._connect_max_attempts_override = 1 + with pytest.raises(OWNConnectionError): + await cmd_sess.connect(raise_on_error=True) + cmd_sess._connect_max_attempts_override = None + with pytest.raises(OWNConnectionError): + await cmd_sess.send("*1*1*11##", raise_on_error=True) + # Second call hits SEND_RECONNECT_COOLDOWN with raise_on_error=True + with pytest.raises(OWNConnectionError): + await cmd_sess.send("*1*1*12##", raise_on_error=True) + + # send_to_gateway(raise_on_error=True) + gw_obj = OWNGateway({"address": "127.0.0.1", "port": 20000}) + with ( + patch.object(OWNCommandSession, "connect", new=AsyncMock(return_value={"Success": True})) as m_conn, + patch.object(OWNCommandSession, "send", new=AsyncMock(return_value=True)) as m_send, + ): + assert await OWNCommandSession.send_to_gateway("*1*1*11##", gw_obj, raise_on_error=True) is True + m_conn.assert_awaited_once_with(raise_on_error=True) + m_send.assert_awaited_once_with("*1*1*11##", raise_on_error=True) + + # send(raise_on_error=True) on unexpected signaling, connection lost, and generic crash + s_proto, _ = make_session(OWNCommandSession) + assert isinstance(s_proto, OWNCommandSession) + s_proto._read_command_response = AsyncMock(return_value=(OWNSignaling("*#*2##"), [])) # type: ignore[method-assign] + with pytest.raises(OWNProtocolError): + await s_proto.send("*1*1*11##", raise_on_error=True) + + s_lost, _ = make_session(OWNCommandSession) + assert isinstance(s_lost, OWNCommandSession) + assert s_lost._stream_reader is not None + s_lost._stream_reader.feed_eof() + with pytest.raises(OWNConnectionError) as lost_exc: + await s_lost.send("*1*1*11##", raise_on_error=True) + assert lost_exc.value.reason == "connection_lost" + + s_crash, _ = make_session(OWNCommandSession) + assert isinstance(s_crash, OWNCommandSession) + s_crash._read_command_response = AsyncMock(side_effect=RuntimeError("boom")) # type: ignore[method-assign] + with pytest.raises(OWNConnectionError) as crash_exc: + await s_crash.send("*1*1*11##", raise_on_error=True) + assert crash_exc.value.reason == "session_crash" + + # Event session _maybe_raise_auth_error when _reconnect returns dict without setting _last_error + ev_sess, _ = make_session(OWNEventSession) + assert isinstance(ev_sess, OWNEventSession) + ev_sess._stream_reader = None + ev_sess._stream_writer = None + with patch.object( + ev_sess, + "_reconnect", + new=AsyncMock(return_value={"Success": False, "Message": "password_error"}), + ): + with pytest.raises(OWNAuthError): + await ev_sess.get_next(raise_on_auth_error=True) + + # Legacy patch("OWNd.connection.find_gateways") compatibility on OWNGateway + with patch("OWNd.connection.find_gateways", new=AsyncMock(return_value=[{"address": "192.168.1.77"}])): + gw = await OWNGateway.get_first_available_gateway() + assert gw is not None and gw.host == "192.168.1.77" +