diff --git a/CHANGELOG.md b/CHANGELOG.md index 51a99e0..655b70d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,13 @@ All notable changes to `OWNd` are documented in this file. - Canonical `libqtdevices` (`TS10_1_0_23`) coarse `WHAT 2..10` $\leftrightarrow$ fine `1..100%` conversion helpers (`COARSE_PRESET_TO_PERCENT`, `brightness_preset_to_percent`, `percent_to_brightness_preset`), `OWNLightingEvent.is_coarse_dimmer`, `OWNLightingEvent.brightness_preset_percent`, and `OWNLightingCommand.set_brightness(..., coarse=..., gateway=...)` (`#134`). - `WHAT 39` end of movement (`*1*39*WHERE##`) from WHO 1 motion detectors: `message_type` `motion_detected` with `motion` `False`, new `OWNLightingEvent.motion_end`, `is_sensor` `True`, `is_on` `None`. It was `unknown_state` `39` before (`MyHOME#777`). +### Changed +- `OWNTransport.send()` takes the same `raise_on_error` keyword as `OWNCommandSession.send()` on `AsyncTcpTransport` and `AsyncSerialTransport`, and `OWNTransport.last_error` holds the error recorded by the last `send()`: a local refusal on both transports, and on the TCP transport everything its command session records (NACK, timeout, connection errors). Both transports now share one contract: `RuntimeError` when not connected, then a frame refused locally is logged at ERROR, kept in `last_error` and answered with `None`, or raised as `OWNProtocolError` with `raise_on_error=True`. + +### Fixed +- **WHO 4 Dimension 22 is never sent**: Dimension 22 is the published split-unit IR control (`*#4*3#Z#N*22##` / `*#4*3#Z#N*#22*MOD*SP*VEL*SWING##`, WHO 4 specification pp. 66-68); in any spelling (`*#4*ZONE*22##`, `*#4*#0*22##`, leading zeros, whitespace or `+` tokens) it crashes the thermoregulation daemon (`bt_termo`) on Linux IP gateways and reboots an F461 2.0.11 about three minutes later (own-firmware-oracle `tools/expected_divergences.yaml` crash frames, live F461 2.0.11). `OWNCommandSession.send` (and so `AsyncTcpTransport.send` and `keepalive`), `OWNCommandSession.send_to_gateway` (before `connect()`: no socket, no handshake) and `AsyncSerialTransport.send` refuse it before anything is written, with the contract above and `reason="destructive_frame"`. The detector `is_who4_dimension_22()` is public, accepts `bytes`, and is anchored to the first `*` of each frame, so a mid-frame WHERE `#4` / DIMENSION `#4` such as `*#2*#4*#11#001*22##` or `*#1*14*#4*1*22##` stays sendable. Parsing is unchanged: the frame can still be read from the bus. +- **`AsyncSerialTransport.send` validates outbound frames** like `OWNCommandSession.send`: an invalid or signaling frame (for example the single-`#` concatenation `*1*1*11#*#4*2*22##`) is refused with `reason="invalid_frame"` instead of being written to the port. The frame is serialised once, terminator included: the string checked is the string written. + ## [2.0.0b11] - 2026-10-20 ### Added diff --git a/OWNd/connection/command_session.py b/OWNd/connection/command_session.py index d8f50b1..7cff5d5 100644 --- a/OWNd/connection/command_session.py +++ b/OWNd/connection/command_session.py @@ -25,7 +25,24 @@ OWNProtocolError, OWNTimeoutError, ) -from ..message import OWNMessage, OWNSignaling +from ..message import OWNMessage, OWNSignaling, is_who4_dimension_22 + +# Why every send path refuses WHO 4 dimension 22 (see +# ``OWNd.message.is_who4_dimension_22``): the gateway never gets the frame. +DESTRUCTIVE_FRAME_REASON = ( + "WHO 4 dimension 22 crashes the thermoregulation daemon (bt_termo) " + "on Linux IP gateways" +) + + +def destructive_frame_error(frame: str) -> OWNProtocolError: + """The error every send path raises or records for a WHO 4 dimension 22 frame.""" + return OWNProtocolError( + f"Refusing to send `{frame}`: {DESTRUCTIVE_FRAME_REASON}", + frame=frame, + reason="destructive_frame", + ) + # Cooldown window (seconds) after an in-send() reconnect attempt fails, so # concurrent callers queued behind _send_lock fail fast instead of each @@ -69,6 +86,15 @@ async def send_to_gateway( raise_on_error: bool = False, ) -> list[OWNMessage | str] | bool | None: connection = cls(gateway) + # Refused before connect(): no socket, no handshake, no command slot + # on a single-session gateway (MH200N, F453AV...) and no 5-attempt + # backoff on an unreachable host for a frame that is never written. + frame = str(message) + if is_who4_dimension_22(frame): + connection._refuse_destructive_frame( + frame, message, raise_on_error=raise_on_error + ) + return None try: if raise_on_error: await connection.connect(raise_on_error=True) @@ -219,6 +245,26 @@ async def send( message, is_status_request, raise_on_error=raise_on_error ) + def _refuse_destructive_frame( + self, frame: str, message: str | OWNMessage, *, raise_on_error: bool + ) -> None: + """Record and log the refusal of a WHO 4 dimension 22 frame, never written. + + ``frame`` is the serialised string the error carries; ``message`` is + the caller's object, passed to the logger like every other send log + line (formatted lazily, and what downstream tests assert on). + """ + err = destructive_frame_error(frame) + self._last_error = err + self._logger.error( + "%s Refusing to send `%s`: %s.", + self._log_id, + message, + DESTRUCTIVE_FRAME_REASON, + ) + if raise_on_error: + raise err + async def _locked_send( self, message: str | OWNMessage, @@ -227,10 +273,16 @@ async def _locked_send( raise_on_error: bool = False, ) -> list[OWNMessage | str] | bool | None: self._last_error = None + # Serialised once: the frame checked is the frame parsed, raised and + # written. Log lines keep the caller's object (formatted lazily). + frame = str(message) + if is_who4_dimension_22(frame): + self._refuse_destructive_frame(frame, message, raise_on_error=raise_on_error) + return None msg_obj = ( message if isinstance(message, OWNMessage) - else OWNMessage.parse(str(message), gateway=self._gateway) + else OWNMessage.parse(frame, gateway=self._gateway) ) is_valid_frame = ( msg_obj is not None @@ -239,8 +291,8 @@ async def _locked_send( ) if not is_valid_frame: err: OWNError = OWNProtocolError( - f"Invalid OpenWebNet frame `{message}`", - frame=str(message), + f"Invalid OpenWebNet frame `{frame}`", + frame=frame, reason="invalid_frame", ) self._last_error = err @@ -292,7 +344,7 @@ async def _locked_send( < SEND_RECONNECT_COOLDOWN ): err = self._last_error or OWNConnectionError( - f"Command session unavailable; message `{message}` not sent", + f"Command session unavailable; message `{frame}` not sent", reason="connection_unavailable", ) self._last_error = err @@ -319,7 +371,7 @@ async def _locked_send( ): self._last_reconnect_failure = time.monotonic() err = self._last_error or OWNConnectionError( - f"Command session unavailable; message `{message}` not sent", + f"Command session unavailable; message `{frame}` not sent", reason="connection_unavailable", ) await self.close() @@ -336,7 +388,7 @@ async def _locked_send( written = False try: - self._stream_writer.write(str(message).encode()) + self._stream_writer.write(frame.encode()) written = True await self._stream_writer.drain() @@ -368,8 +420,8 @@ async def _locked_send( message, ) err = OWNNackError( - f"Gateway rejected message `{message}` with NACK", - frame=str(message), + f"Gateway rejected message `{frame}` with NACK", + frame=frame, responses=list(collected), ) self._last_error = err @@ -391,7 +443,7 @@ async def _locked_send( message, ) err = OWNProtocolError( - f"Unexpected response `{resulting_message}` to message `{message}`", + f"Unexpected response `{resulting_message}` to message `{frame}`", frame=str(resulting_message), reason="unexpected_response", ) @@ -408,7 +460,7 @@ async def _locked_send( except TimeoutError as error: await self.close() err = OWNTimeoutError( - f"Timed out awaiting the complete response for `{message}`", + f"Timed out awaiting the complete response for `{frame}`", reason="command_timeout", ) self._last_error = err @@ -441,7 +493,7 @@ async def _locked_send( # 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})", + f"Connection lost before acknowledgement of `{frame}` ({error})", reason="connection_lost", ) self._last_error = err diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index 99a2280..ac0d6a1 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -171,6 +171,7 @@ _ZONE_STATES, _zone_state, _zone_state_text, + is_who4_dimension_22, who4_temperature, OWNHeatingCommand, OWNHeatingEvent, @@ -314,6 +315,7 @@ "who4_temperature", "_zone_state", "_zone_state_text", + "is_who4_dimension_22", "OWNHeatingEvent", "OWNHeatingCommand", # Burglar Alarm (WHO 5) diff --git a/OWNd/message/heating.py b/OWNd/message/heating.py index 74fd51e..a9a38f9 100644 --- a/OWNd/message/heating.py +++ b/OWNd/message/heating.py @@ -74,6 +74,40 @@ _WHO4_TEMP_REGEX = re.compile(r"^[01][0-9]{3}$") +# WHO 4 DIMENSION 22 is the published split-unit (air conditioning) IR +# control: ``*#4*3#Z#N*22##`` requests the current setting and +# ``*#4*3#Z#N*#22*MOD*SP*VEL*SWING##`` writes it (WHO 4 specification, +# pp. 66-68). On Linux IP gateways the frame kills the thermoregulation +# daemon (bt_termo) and the gateway reboots about 3 minutes later (firmware +# oracle crash_frames; live F461 2.0.11). Every send path refuses it, in any +# shape: request, write (``*#22*``) or reply, with or without a dimension +# parameter, and in any ``*#4*ZONE*22##`` spelling. +# +# Anchored to the first ``*`` of each ``##``-delimited frame: ``*#4*`` means +# WHO 4 only there. Mid-frame, ``*#4*`` is WHERE ``#4`` (group 4 in WHO 1, 2, +# 16 and 22, central-unit zone 4 in WHO 4) or DIMENSION ``#4`` (a dimension 4 +# write), so ``*#2*#4*#11#001*22##`` or ``*#1*14*#4*1*22##`` must not match. +# The WHO and DIMENSION tokens accept whatever the gateway's atoi() reads as +# 4 / 22: leading zeros, surrounding whitespace and a ``+`` sign +# (``*#04*2* 22##``, ``*#4*2*+22##``, ``*#4*2*22 ``). +_WHO4_DIMENSION_22_REGEX = re.compile( + r"^[^*]*\*#\s*\+?0*4\s*\*[^*]*\*\s*#?\s*\+?0*22(?:\s*$|[#*\s])" +) + + +def is_who4_dimension_22(frame: object) -> bool: + """True if ``frame`` (or any ``##``-delimited frame in it) is WHO 4 DIMENSION 22. + + ``bytes``, ``bytearray`` and ``memoryview`` are decoded as latin-1 (the + wire encoding); anything else is checked through ``str()``. + """ + raw = ( + bytes(frame).decode("latin-1") + if isinstance(frame, (bytes, bytearray, memoryview)) + else str(frame) + ) + return any(_WHO4_DIMENSION_22_REGEX.search(chunk) for chunk in raw.split("##")) + def who4_temperature(raw: str) -> float | None: """Decode WHO 4 temperature ``SXXX`` (tenths of a degree).""" diff --git a/OWNd/transport/base.py b/OWNd/transport/base.py index e98da9d..cae1a48 100644 --- a/OWNd/transport/base.py +++ b/OWNd/transport/base.py @@ -7,6 +7,7 @@ from collections.abc import Callable from typing import Any +from ..exceptions import OWNError from ..message import OWNMessage Listener = Callable[[OWNMessage | str], Any] @@ -23,6 +24,18 @@ def __init__( self.log_id = log_id self._logger = logger or logging.getLogger(__name__) self._listeners: set[Listener] = set() + self._last_error: OWNError | None = None + + @property + def last_error(self) -> OWNError | None: + """The error recorded by the last ``send()``, if any. + + Both transports record a local refusal (``OWNProtocolError``); the + TCP transport exposes its command session's ``last_error``, which + also covers NACK, timeout and connection errors. A serial NACK or + timeout returns ``None`` without recording an error. + """ + return self._last_error @property @abstractmethod @@ -44,9 +57,21 @@ async def disconnect(self) -> None: @abstractmethod 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 one command and return its response.""" + """Send one command and return its response. + + Raises ``RuntimeError`` when the transport is not connected. A frame + that is refused locally (invalid, or WHO 4 dimension 22 which crashes + the gateway's thermoregulation daemon) is logged at ERROR, kept in + ``last_error`` and answered with ``None``; with ``raise_on_error`` it + raises ``OWNProtocolError`` (``reason`` ``invalid_frame`` or + ``destructive_frame``) instead, like ``OWNCommandSession.send``. + """ def register_listener(self, callback: Listener) -> Callable[[], None]: """Register an inbound-message callback and return its unsubscribe.""" diff --git a/OWNd/transport/serial.py b/OWNd/transport/serial.py index 27e06e8..b962427 100644 --- a/OWNd/transport/serial.py +++ b/OWNd/transport/serial.py @@ -7,7 +7,12 @@ import logging from ..connection import COMMAND_RESPONSE_MAX_FRAMES, COMMAND_TIMEOUT -from ..message import OWNMessage, OWNSignaling +from ..connection.command_session import ( + DESTRUCTIVE_FRAME_REASON, + destructive_frame_error, +) +from ..exceptions import OWNError, OWNProtocolError +from ..message import OWNMessage, OWNSignaling, is_who4_dimension_22 from .base import OWNTransport SEPARATOR = b"##" @@ -142,24 +147,67 @@ def _process_frame(self, frame: str) -> None: self._pending_frames.append(message) 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: del is_status_request if not self._connected or self._writer is None: raise RuntimeError("Serial transport is not connected") + self._last_error = None + + # Serialised once, terminator included: the frame checked is the + # frame parsed, logged and written. + text = str(message) + if not text.endswith("##"): + text += "##" + if is_who4_dimension_22(text): + # Never written, whatever the caller: it reboots the gateway. + err: OWNError = destructive_frame_error(text) + self._last_error = err + self._logger.error( + "%s Refusing to send `%s`: %s.", + self.log_id, + text, + DESTRUCTIVE_FRAME_REASON, + ) + if raise_on_error: + raise err + return None + msg_obj = ( + message + if isinstance(message, OWNMessage) + else OWNMessage.parse(text, gateway=self.port) + ) + if ( + msg_obj is None + or not msg_obj.is_valid + or isinstance(msg_obj, OWNSignaling) + ): + err = OWNProtocolError( + f"Invalid OpenWebNet frame `{text}`", + frame=text, + reason="invalid_frame", + ) + self._last_error = err + self._logger.error( + "%s Refusing to send invalid OpenWebNet frame `%s`.", + self.log_id, + text, + ) + if raise_on_error: + raise err + return None async with self._command_lock: loop = asyncio.get_running_loop() self._pending_future = loop.create_future() - self._pending_request = OWNMessage.parse( - str(message), gateway=self.port - ) + self._pending_request = msg_obj self._pending_frames = [] try: - frame = str(message).encode() - if not frame.endswith(SEPARATOR): - frame += SEPARATOR - self._writer.write(frame) + self._writer.write(text.encode()) await self._writer.drain() return await asyncio.wait_for( self._pending_future, timeout=COMMAND_TIMEOUT @@ -168,7 +216,7 @@ async def send( self._logger.warning( "%s Timed out waiting for serial response to `%s`.", self.log_id, - message, + text, ) return None finally: diff --git a/OWNd/transport/tcp.py b/OWNd/transport/tcp.py index 7b94945..344240b 100644 --- a/OWNd/transport/tcp.py +++ b/OWNd/transport/tcp.py @@ -7,6 +7,7 @@ import logging from ..connection import OWNCommandSession, OWNEventSession, OWNGateway +from ..exceptions import OWNError from ..message import OWNMessage from .base import OWNTransport @@ -37,6 +38,12 @@ def is_connected(self) -> bool: def transport_type(self) -> str: return "tcp" + @property + def last_error(self) -> OWNError | None: + if self._command_session is None: + return None + return self._command_session.last_error + async def connect(self) -> bool: if self._event_session is not None or self._command_session is not None: await self.disconnect() @@ -91,8 +98,14 @@ async def disconnect(self) -> None: await command_session.close() 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: if self._command_session is None: raise RuntimeError("TCP transport is not connected") - return await self._command_session.send(message, is_status_request) + return await self._command_session.send( + message, is_status_request, raise_on_error=raise_on_error + ) diff --git a/tests/test_transport_extended.py b/tests/test_transport_extended.py index c708020..96d78bb 100644 --- a/tests/test_transport_extended.py +++ b/tests/test_transport_extended.py @@ -174,7 +174,9 @@ async def fake_get_next(): # Send command res = await transport.send("*1*1*12##", is_status_request=False) assert res is True - mock_cmd_sess.send.assert_called_once_with("*1*1*12##", False) + mock_cmd_sess.send.assert_called_once_with( + "*1*1*12##", False, raise_on_error=False + ) # Disconnect await transport.disconnect() diff --git a/tests/test_who4_dimension22_guard.py b/tests/test_who4_dimension22_guard.py new file mode 100644 index 0000000..546a776 --- /dev/null +++ b/tests/test_who4_dimension22_guard.py @@ -0,0 +1,522 @@ +"""WHO 4 DIMENSION 22 never leaves OWNd. + +``*#4*ZONE*22##`` kills the gateway's thermoregulation daemon (bt_termo) and +reboots an F461 about 3 minutes later (firmware oracle crash_frames; live F461 +2.0.11). The frames below are test inputs only: every send path must refuse +them before anything is written. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from OWNd.connection import OWNCommandSession, OWNGateway +from OWNd.exceptions import OWNNackError, OWNProtocolError +from OWNd.message import ( + OWNAutomationCommand, + OWNCommand, + OWNHeatingCommand, + OWNLightingCommand, + OWNMessage, + OWNSoundCommand, + OWNSoundDiffusionCommand, + is_who4_dimension_22, +) +from OWNd.transport.serial import AsyncSerialTransport +from OWNd.transport.tcp import AsyncTcpTransport + +DIMENSION_22_FRAMES = ( + "*#4*2*22##", + "*#4*#0*22##", + "*#4*#1*22##", + "*#4*2#1*22##", + "*#4**22##", + "*#4*2*22#1##", + "*#4*2*#22*1##", + "*#4*2*22*1##", + "*#04*2*022##", + "*1*1*11##*#4*2*22##", + # Published split-unit forms (WHO 4 specification, pp. 66-68). + "*#4*3#2#1*22##", + "*#4*3#2#1*#22*0***##", + "*#4*45#1*#22*3**1*1##", + # Tokens the gateway's atoi() still reads as 4 / 22: whitespace, sign, + # missing terminator (the serial transport appends ``##``). + "*#4*2*22 ", + "*#4*2*22 ##", + "*#4*2* 22##", + "*#4*2*+22##", + "*# 4*2*22##", + "*#+4*2*22##", + "*#4*2*22", +) + +# Zone 22, other dimensions, dimension 22 of other WHOs, and mid-frame +# ``*#4*`` tokens (WHERE ``#4`` or DIMENSION ``#4``) stay sendable. +SAFE_FRAMES = ( + "*#4*22##", + "*#4*22*0##", + "*#4*#22*14##", + "*#4*2*20##", + "*#4*2*220##", + "*#4*2*0*0220##", + "*#4*2*2 2##", + "*#4*2*-22##", + "*#44*2*22##", + "*#13**22##", + "*#1*22*1##", + "*4*303*22##", + # WHERE ``#4`` (group 4 / central-unit zone 4) with a value of 22. + "*#2*#4*#11#001*22##", + "*#1*#4*#2*22*0*0##", + "*#1*#4*#12*22*100*50##", + "*#16*#4*#1*22##", + "*#22*#4*#1*22##", + "*#22*#4*#19*22##", + "*#4*#4*#30*22*10*2026##", + "*#4*#4*#31*22*30##", + "*#4*#4*#32*22*00##", + # DIMENSION ``#4`` write whose second value is 22. + "*#1*14*#4*1*22##", + # Other subsystems carrying a 22. + "*#1004*2*22##", + "*#22*3#1#1*1##", + # Single-``#`` concatenation: one chunk starting with ``*1*``; the frame + # validation of every send path refuses it as invalid instead. + "*1*1*11#*#4*2*22##", +) + +# Builders whose WHERE is group / central-unit zone ``#4`` and whose first +# value is 22: the unanchored detector used to refuse all of them. +WHERE_HASH_4_BUILDERS = ( + (lambda: OWNAutomationCommand.set_shutter_level("#4", 22), "*#2*#4*#11#001*22##"), + (lambda: OWNLightingCommand.set_variable_timer("#4", 22, 0, 0), "*#1*#4*#2*22*0*0##"), + ( + lambda: OWNLightingCommand.set_hsv_color("#4", 22, 100, 50), + "*#1*#4*#12*22*100*50##", + ), + (lambda: OWNSoundCommand.set_volume("#4", 22), "*#16*#4*#1*22##"), + (lambda: OWNSoundDiffusionCommand.set_volume("#4", 22), "*#22*#4*#1*22##"), + (lambda: OWNSoundDiffusionCommand.set_preset("#4", 22), "*#22*#4*#6*22##"), + ( + lambda: OWNHeatingCommand.set_holiday_end_date("#4", 22, 10, 2026), + "*#4*#4*#30*22*10*2026##", + ), + (lambda: OWNHeatingCommand.set_holiday_end_time("#4", 22, 30), "*#4*#4*#31*22*30##"), + ( + lambda: OWNHeatingCommand.set_timed_manual_end_time("#4", 22, 0), + "*#4*#4*#32*22*00##", + ), +) + + +class _Writer: + def __init__(self) -> None: + self.written: list[bytes] = [] + + def write(self, data: bytes) -> None: + self.written.append(data) + + async def drain(self) -> None: + return None + + def close(self) -> None: + return None + + async def wait_closed(self) -> None: + return None + + +def _command_session() -> tuple[OWNCommandSession, _Writer]: + gateway = OWNGateway({"address": "192.0.2.1", "port": 20000, "password": "12345"}) + session = OWNCommandSession(gateway=gateway) + session._stream_reader = asyncio.StreamReader() + writer = _Writer() + session._stream_writer = writer # type: ignore[assignment] + session._read_frame = AsyncMock(return_value="*#*1##") # type: ignore[method-assign] + return session, writer + + +@pytest.mark.parametrize("frame", DIMENSION_22_FRAMES) +def test_detector_flags_every_dimension_22_form(frame: str) -> None: + assert is_who4_dimension_22(frame) is True + + +@pytest.mark.parametrize("frame", SAFE_FRAMES) +def test_detector_leaves_other_frames_alone(frame: str) -> None: + assert is_who4_dimension_22(frame) is False + + +@pytest.mark.parametrize( + "frame", + [b"*#4*2*22##", bytearray(b"*#4*2*22##"), memoryview(b"*#4*2*22##")], + ids=["bytes", "bytearray", "memoryview"], +) +def test_detector_decodes_buffers(frame: object) -> None: + assert is_who4_dimension_22(frame) is True + assert is_who4_dimension_22(b"*#4*#4*#30*22*10*2026##") is False + + +@pytest.mark.parametrize( + "build, expected", + WHERE_HASH_4_BUILDERS, + ids=[expected for _, expected in WHERE_HASH_4_BUILDERS], +) +def test_detector_leaves_where_hash_4_builders_alone( + build: Callable[[], OWNCommand], expected: str +) -> None: + command = build() + assert str(command) == expected + assert command.is_valid + assert is_who4_dimension_22(command) is False + assert is_who4_dimension_22(str(command)) is False + + +@pytest.mark.parametrize( + "build, expected", + WHERE_HASH_4_BUILDERS, + ids=[expected for _, expected in WHERE_HASH_4_BUILDERS], +) +@pytest.mark.asyncio +async def test_command_session_still_sends_where_hash_4_builders( + build: Callable[[], OWNCommand], expected: str +) -> None: + session, writer = _command_session() + + assert await session.send(build()) is True + assert writer.written == [expected.encode()] + + +@pytest.mark.parametrize("frame", DIMENSION_22_FRAMES) +@pytest.mark.asyncio +async def test_command_session_refuses_raw_string(frame: str) -> None: + session, writer = _command_session() + + assert await session.send(frame) is None + assert writer.written == [] + assert isinstance(session._last_error, OWNProtocolError) + assert session._last_error.reason == "destructive_frame" + + with pytest.raises(OWNProtocolError) as excinfo: + await session.send(frame, is_status_request=True, raise_on_error=True) + assert excinfo.value.reason == "destructive_frame" + assert writer.written == [] + + +@pytest.mark.parametrize( + "message", + [ + OWNCommand.parse("*#4*2*22##"), + OWNMessage.parse("*#4*#0*22##"), + OWNHeatingCommand("*#4*2#1*22##"), + ], + ids=["OWNCommand.parse", "OWNMessage.parse", "OWNHeatingCommand"], +) +@pytest.mark.asyncio +async def test_command_session_refuses_parsed_and_built_messages(message: OWNMessage) -> None: + # The parser still reads the frame (it may appear on the bus); only sending is refused. + assert isinstance(message, OWNHeatingCommand) + assert message.is_valid + session, writer = _command_session() + + with pytest.raises(OWNProtocolError, match="dimension 22"): + await session.send(message, raise_on_error=True) + assert writer.written == [] + + +class _CountingCommand(OWNHeatingCommand): + """Counts serialisations: the string checked must be the one parsed, raised and written.""" + + def __init__(self, raw: str) -> None: + super().__init__(raw) + self.str_calls = 0 + + def __str__(self) -> str: + self.str_calls += 1 + return super().__str__() + + +@pytest.mark.parametrize( + "raw, reply, expected, writes", + [ + ("*#4*22##", "*#*1##", True, 1), + ("*#4*22##", "*#*0##", None, 2), # a NACKed command is retried once + ("*#4*2*22##", "*#*1##", None, 0), + ], + ids=["ack", "nack", "destructive"], +) +@pytest.mark.asyncio +async def test_command_session_serialises_once( + raw: str, reply: str, expected: bool | None, writes: int +) -> None: + session, writer = _command_session() + session._read_frame = AsyncMock(return_value=reply) # type: ignore[method-assign] + # Log lines receive the message object and format it lazily (MyHOME's + # suite asserts on those call arguments), so the logger is a mock here + # and the count covers the code path only. + session._logger = MagicMock() + message = _CountingCommand(raw) + + assert await session.send(message) is expected + assert message.str_calls == 1 + assert writer.written == [raw.encode()] * writes + if expected is None: + assert isinstance(session.last_error, OWNProtocolError | OWNNackError) + assert session.last_error.frame == raw + + +@pytest.mark.asyncio +async def test_nack_contract_is_unchanged() -> None: + """The NACK path keeps its observable behaviour: result, retries, logging. + + MyHOME's suite (run by the conformance gate against this checkout) asserts + the DEBUG / ERROR log calls with the message object, not its string. + """ + session, writer = _command_session() + session._read_frame = AsyncMock(return_value="*#*0##") # type: ignore[method-assign] + session._logger = MagicMock() + + # A NACKed status request is final: one write, DEBUG only, None. + status = OWNCommand.parse("*#16*0*5##") + assert await session.send(status, is_status_request=True) is None + assert writer.written == [b"*#16*0*5##"] + assert isinstance(session.last_error, OWNNackError) + assert session.last_error.frame == "*#16*0*5##" + session._logger.debug.assert_any_call( + "%s Gateway rejected status request %s (NACK, %s response(s)). Subsystem or device may not be present.", + session._log_id, + status, + 0, + ) + session._logger.warning.assert_not_called() + session._logger.error.assert_not_called() + + # A NACKed command is retried once, then reported at ERROR with the object. + writer.written.clear() + command = OWNCommand.parse("*1*1*21##") + assert await session.send(command) is None + assert writer.written == [b"*1*1*21##", b"*1*1*21##"] + session._logger.error.assert_called_with( + "%s Could not send message `%s`. No more retries.", session._log_id, command + ) + with pytest.raises(OWNNackError): + await session.send(command, raise_on_error=True) + + +@pytest.mark.asyncio +async def test_command_session_refuses_before_reconnecting() -> None: + session, _ = _command_session() + session._stream_writer = None + session._stream_reader = None + with patch.object(session, "connect", new_callable=AsyncMock) as connect: + assert await session.send("*#4*2*22##") is None + connect.assert_not_called() + + +@pytest.mark.parametrize( + "message", + ["*#4*2*22##", OWNHeatingCommand("*#4*3#2#1*22##")], + ids=["str", "OWNHeatingCommand"], +) +@pytest.mark.asyncio +async def test_send_to_gateway_refuses_without_connecting( + message: str | OWNMessage, caplog: pytest.LogCaptureFixture +) -> None: + # Fail fast: no socket, no handshake, no gateway command slot, no + # 5-attempt backoff on an unreachable host. + gateway = OWNGateway({"address": "192.0.2.1", "port": 20000, "password": "12345"}) + + with ( + patch.object(OWNCommandSession, "connect", new_callable=AsyncMock) as connect, + patch.object(OWNCommandSession, "_locked_send", new_callable=AsyncMock) as send, + caplog.at_level(logging.ERROR), + ): + with pytest.raises(OWNProtocolError) as excinfo: + await OWNCommandSession.send_to_gateway(message, gateway, raise_on_error=True) + assert excinfo.value.reason == "destructive_frame" + assert excinfo.value.frame == str(message) + assert await OWNCommandSession.send_to_gateway(message, gateway) is None + connect.assert_not_called() + send.assert_not_called() + assert caplog.text.count("bt_termo") == 2 + + +@pytest.mark.asyncio +async def test_send_to_gateway_still_connects_for_other_frames() -> None: + gateway = OWNGateway({"address": "192.0.2.1", "port": 20000, "password": "12345"}) + + with ( + patch.object( + OWNCommandSession, + "connect", + new_callable=AsyncMock, + return_value={"Success": True}, + ) as connect, + patch.object( + OWNCommandSession, "_locked_send", new_callable=AsyncMock, return_value=True + ) as send, + ): + assert await OWNCommandSession.send_to_gateway("*#4*22##", gateway) is True + connect.assert_awaited_once() + send.assert_awaited_once_with("*#4*22##", False, raise_on_error=False) + + +@pytest.mark.parametrize("frame", SAFE_FRAMES[:3]) +@pytest.mark.asyncio +async def test_command_session_still_sends_zone_22(frame: str) -> None: + session, writer = _command_session() + + assert await session.send(frame, is_status_request=True) is True + assert writer.written == [frame.encode()] + + +def _serial_transport() -> tuple[AsyncSerialTransport, _Writer]: + transport = AsyncSerialTransport(port="/dev/null") + writer = _Writer() + transport._writer = writer # type: ignore[assignment] + transport._connected = True + return transport, writer + + +@pytest.mark.asyncio +async def test_tcp_transport_refuses_through_its_command_session( + caplog: pytest.LogCaptureFixture, +) -> None: + gateway = OWNGateway({"address": "192.0.2.1", "port": 20000, "password": "12345"}) + transport = AsyncTcpTransport(gateway=gateway) + assert transport.last_error is None + session, writer = _command_session() + transport._command_session = session + + with caplog.at_level(logging.ERROR): + assert await transport.send("*#4*2*22##") is None + assert writer.written == [] + assert transport.last_error is session.last_error + assert isinstance(transport.last_error, OWNProtocolError) + assert transport.last_error.reason == "destructive_frame" + assert "bt_termo" in caplog.text + + # The keyword reaches the command session: same contract on both transports. + with pytest.raises(OWNProtocolError) as excinfo: + await transport.send("*#4*2*22##", raise_on_error=True) + assert excinfo.value.reason == "destructive_frame" + assert writer.written == [] + + +@pytest.mark.parametrize("frame", DIMENSION_22_FRAMES) +@pytest.mark.asyncio +async def test_serial_transport_refuses( + frame: str, caplog: pytest.LogCaptureFixture +) -> None: + transport, writer = _serial_transport() + + # Default: like AsyncTcpTransport, log at ERROR, keep the error, return None. + with caplog.at_level(logging.ERROR): + assert await transport.send(frame) is None + assert writer.written == [] + assert transport._pending_future is None + assert isinstance(transport.last_error, OWNProtocolError) + assert transport.last_error.reason == "destructive_frame" + assert transport.last_error.frame.endswith("##") + assert "bt_termo" in caplog.text + + with pytest.raises(OWNProtocolError) as excinfo: + await transport.send(frame, raise_on_error=True) + assert excinfo.value.reason == "destructive_frame" + assert excinfo.value.frame == transport.last_error.frame + assert writer.written == [] + assert transport._pending_future is None + + +@pytest.mark.asyncio +async def test_serial_transport_refuses_built_message() -> None: + transport, writer = _serial_transport() + + with pytest.raises(OWNProtocolError, match="dimension 22"): + await transport.send(OWNHeatingCommand("*#4*2#1*22##"), raise_on_error=True) + assert writer.written == [] + + +@pytest.mark.parametrize( + "frame", + ["*1*1*11#*#4*2*22##", "*#*1##", "not a frame", ""], + ids=["single-hash-concatenation", "signaling", "text", "empty"], +) +@pytest.mark.asyncio +async def test_serial_transport_refuses_invalid_frames( + frame: str, caplog: pytest.LogCaptureFixture +) -> None: + # Like OWNCommandSession._locked_send: only a valid, non-signaling frame is written. + transport, writer = _serial_transport() + + with caplog.at_level(logging.ERROR): + assert await transport.send(frame) is None + assert writer.written == [] + assert transport._pending_future is None + assert isinstance(transport.last_error, OWNProtocolError) + assert transport.last_error.reason == "invalid_frame" + assert "invalid OpenWebNet frame" in caplog.text + + with pytest.raises(OWNProtocolError) as excinfo: + await transport.send(frame, raise_on_error=True) + assert excinfo.value.reason == "invalid_frame" + assert writer.written == [] + + +# Spellings a C atoi() would still read as WHO 4 / dimension 22 but that are +# not OpenWebNet tokens: the detector leaves them to the frame validation, +# which refuses them as invalid on both paths, so they are never written. +ATOI_PREFIX_FRAMES = ("*#4junk*2*22##", "*#4*2*22junk##", "*#4.0*2*22##", "*#4*2*22.5##") + + +@pytest.mark.parametrize("frame", ATOI_PREFIX_FRAMES) +@pytest.mark.asyncio +async def test_atoi_prefix_spellings_are_refused_as_invalid(frame: str) -> None: + assert OWNMessage.parse(frame) is None + + session, session_writer = _command_session() + assert await session.send(frame) is None + assert isinstance(session.last_error, OWNProtocolError) + assert session.last_error.reason == "invalid_frame" + assert session_writer.written == [] + + transport, serial_writer = _serial_transport() + with pytest.raises(OWNProtocolError) as excinfo: + await transport.send(frame, raise_on_error=True) + assert excinfo.value.reason == "invalid_frame" + assert serial_writer.written == [] + + +@pytest.mark.asyncio +async def test_serial_transport_not_connected_comes_first() -> None: + # Same ordering as AsyncTcpTransport: an unconnected transport is a + # RuntimeError whatever the frame; the frame checks run on a connected one. + transport = AsyncSerialTransport(port="/dev/null") + with pytest.raises(RuntimeError, match="not connected"): + await transport.send("*#4*2*22##", raise_on_error=True) + assert transport.last_error is None + + +@pytest.mark.asyncio +async def test_serial_transport_still_sends_safe_frames() -> None: + transport, writer = _serial_transport() + assert transport.last_error is None + + # The third frame has no terminator: the serial transport appends it. + for frame in ("*#4*22##", "*#1*#4*#2*22*0*0##", "*#4*#4*#30*22*10*2026"): + + async def ack() -> None: + await asyncio.sleep(0) + assert transport._pending_future is not None + transport._pending_future.set_result(True) + + task = asyncio.create_task(ack()) + assert await transport.send(frame, raise_on_error=True) is True + await task + assert transport.last_error is None + assert writer.written == [b"*#4*22##", b"*#1*#4*#2*22*0*0##", b"*#4*#4*#30*22*10*2026##"] diff --git a/tests/typing/consumer_send.py b/tests/typing/consumer_send.py index 3432c0a..3534856 100644 --- a/tests/typing/consumer_send.py +++ b/tests/typing/consumer_send.py @@ -70,3 +70,7 @@ async def send_via_transport( parsed = OWNMessage.parse(frame) if parsed is not None: await transport.send(parsed) + # Every transport takes the same keyword as OWNCommandSession.send. + await transport.send("*1*1*11##", raise_on_error=True) + await tcp_transport.send(OWNLightingCommand.switch_on("11"), raise_on_error=True) + await serial_transport.send("*1*1*11##", is_status_request=False, raise_on_error=True)