From 1e4b042b6b18c6ac061b599f2f6712e380c4eb32 Mon Sep 17 00:00:00 2001 From: botts7 <46076879+botts7@users.noreply.github.com> Date: Wed, 29 Apr 2026 23:51:53 +1000 Subject: [PATCH] Add DTLS transport for Wi-Fi/LAN Kit-20 dongles with newer firmware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Newer Kit-20 dongle firmware (released ~2024 onward) encrypts the local Modbus protocol with DTLS over UDP/8899. The standard `WIFIKIT-214028-READ` discovery probe to UDP/48899 returns `dtls_port:8899` for affected hardware, while plaintext requests on UDP/8899 are silently dropped — the existing UdpInverterProtocol times out with no useful diagnostic. This change adds an opt-in DTLS transport that runs the unchanged Modbus RTU framing over DTLS. The application protocol (frames, register layout, slave addresses, AA55 response prefix) is identical to the plaintext path; only the transport differs. Empirically observed handshake parameters: - DTLS 1.2 negotiated by default with OpenSSL 3.0+ defaults (cipher ECDHE-RSA-AES256-GCM-SHA384). DTLS 1.0 also accepted. - Self-signed placeholder certificate, so peer verification is disabled. - Appears to accept only one concurrent DTLS session. Sessions are reused via keep_alive=True so the ~1 s handshake is paid once per Inverter instance. On a stale-response retry the session is preserved and the drain at the start of the next send absorbs any leftover data; only real I/O / SSL exceptions tear the session down. Usage: inverter = await goodwe.connect(ip, family="DT", dtls=True) Requires pyOpenSSL — installed via the new `[dtls]` extra. Auto-discovery remains plaintext-only, so the family must be specified explicitly when dtls=True. Tested on a single-phase MS-family inverter (GW10K-MS-30) running a Kit-20 dongle whose discovery response includes `dtls_port:8899`. Existing 122 tests still pass; 10 new tests cover the DTLS transport's command construction, retry/lock semantics, stale-response handling, and lifecycle. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 19 +++ goodwe/__init__.py | 17 ++- goodwe/dt.py | 4 +- goodwe/dtls.py | 262 +++++++++++++++++++++++++++++++++++++ goodwe/es.py | 4 +- goodwe/et.py | 4 +- goodwe/inverter.py | 11 +- setup.cfg | 3 + tests/manual_dtls_check.py | 48 +++++++ tests/test_dtls.py | 169 ++++++++++++++++++++++++ 10 files changed, 533 insertions(+), 8 deletions(-) create mode 100644 goodwe/dtls.py create mode 100644 tests/manual_dtls_check.py create mode 100644 tests/test_dtls.py diff --git a/README.md b/README.md index 7dc1395..57e9f41 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,25 @@ know to work properly. Since v0.4.x the library also supports standard Modbus/TCP over port 502. This protocol is supported by the V2.0 version of LAN+WiFi communication dongle (model WLA0000-01-00P). +### Newer Wi-Fi/LAN Kit-20 firmware (DTLS) + +Some newer Wi-Fi/LAN Kit-20 dongles (firmware released ~2024 onward — sometimes +marketed as the "Cyber Security dongle") encrypt their local Modbus protocol with +DTLS over UDP/8899. The standard `WIFIKIT-214028-READ` discovery probe to UDP/48899 +returns `dtls_port:8899` for affected hardware; plaintext requests on UDP/8899 are +silently dropped. + +Pass `dtls=True` to `goodwe.connect()` to use the DTLS transport. The application +protocol (Modbus RTU framing, register layout) is unchanged from the plaintext path +— only the transport differs. Auto-discovery is plaintext-only, so the inverter +family must be specified explicitly when `dtls=True`: + +```python +inverter = await goodwe.connect(ip_address, family="DT", dtls=True) +``` + +Requires `pyOpenSSL` — install via `pip install goodwe[dtls]`. + (If you can't communicate with the inverter despite your model is listed above, it is possible you have old ARM firmware version. You should ask manufacturer support to upgrade your ARM firmware (not just inverter firmware) to be able to communicate with the inverter.) diff --git a/goodwe/__init__.py b/goodwe/__init__.py index 13f7ab6..ab353ff 100644 --- a/goodwe/__init__.py +++ b/goodwe/__init__.py @@ -33,6 +33,7 @@ async def connect( timeout: int = 1, retries: int = 3, do_discover: bool = True, + dtls: bool = False, ) -> Inverter: """Contact the inverter at the specified host/port and answer appropriate Inverter instance. @@ -45,15 +46,25 @@ async def connect( Since the UDP communication is by definition unreliable, when no (valid) response is received by the specified timeout, it is considered lost and the command will be re-tried up to retries times. + Newer Wi-Fi/LAN Kit-20 dongles (firmware that advertises ``dtls_port:8899`` in the + discovery probe response on UDP/48899) encrypt local communication with DTLS. Pass + ``dtls=True`` to use the DTLS transport. Requires pyOpenSSL — install via + ``pip install goodwe[dtls]``. Family must be specified explicitly when dtls=True + (auto-discovery is not supported on the encrypted path). + Raise InverterError if unable to contact or recognise supported inverter. """ if family in ET_FAMILY: - inv = ET(host, port, comm_addr, timeout, retries) + inv = ET(host, port, comm_addr, timeout, retries, dtls=dtls) elif family in ES_FAMILY: - inv = ES(host, port, comm_addr, timeout, retries) + inv = ES(host, port, comm_addr, timeout, retries, dtls=dtls) elif family in DT_FAMILY: - inv = DT(host, port, comm_addr, timeout, retries) + inv = DT(host, port, comm_addr, timeout, retries, dtls=dtls) elif do_discover: + if dtls: + raise InverterError( + "dtls=True requires explicit family; auto-discovery uses plaintext" + ) return await discover(host, port, timeout, retries) else: raise InverterError("Specify either an inverter family or set do_discover True") diff --git a/goodwe/dt.py b/goodwe/dt.py index a078d24..0314e44 100644 --- a/goodwe/dt.py +++ b/goodwe/dt.py @@ -210,8 +210,10 @@ def __init__( comm_addr: int = 0, timeout: int = 1, retries: int = 3, + dtls: bool = False, ): - super().__init__(host, port, comm_addr if comm_addr else 0x7F, timeout, retries) + super().__init__(host, port, comm_addr if comm_addr else 0x7F, + timeout, retries, dtls=dtls) self._READ_DEVICE_VERSION_INFO: ProtocolCommand = self._read_command( 0x7531, 0x0028 ) diff --git a/goodwe/dtls.py b/goodwe/dtls.py new file mode 100644 index 0000000..817f115 --- /dev/null +++ b/goodwe/dtls.py @@ -0,0 +1,262 @@ +"""DTLS transport for newer Wi-Fi/LAN Kit-20 dongles. + +Newer Kit-20 firmware (some dongle revisions marketed as "Cyber Security +dongle") moves the local Modbus port from plaintext UDP/8899 to DTLS-protected +UDP/8899. The standard ``WIFIKIT-214028-READ`` discovery probe to UDP/48899 +returns ``dtls_port:8899`` for affected dongles, while the existing +:class:`UdpInverterProtocol` silently times out against them. + +This module adds an opt-in transport that runs the unchanged Modbus RTU +framing over DTLS. Application protocol (frames, register layout, slave +addresses, response prefix) is identical to the plaintext path; only the +transport layer is different. + +Empirically observed handshake parameters: + - DTLS 1.2 negotiated by default with OpenSSL 3.0+ defaults + (cipher ECDHE-RSA-AES256-GCM-SHA384 in our test). DTLS 1.0 also accepted. + - The dongle presents a self-signed placeholder certificate, so peer + verification must be disabled (``SSL_VERIFY_NONE``). + - The dongle accepts only one concurrent DTLS session. + +Sessions are opened lazily on first request and reused (``keep_alive=True``). +On transient UDP loss the session is preserved and the request is retried +once on the same session after draining any stale buffered data; only +persistent failures tear the session down. + +Requires pyOpenSSL — install via ``pip install goodwe[dtls]``. +""" + +from __future__ import annotations + +import asyncio +import logging +import select +import socket +import time +from asyncio.futures import Future +from typing import Optional + +from .exceptions import RequestFailedException +from .protocol import ( + InverterProtocol, + ModbusRtuReadCommand, + ModbusRtuWriteCommand, + ModbusRtuWriteMultiCommand, + ProtocolCommand, +) + +logger = logging.getLogger(__name__) + +try: + from OpenSSL import SSL +except ImportError as exc: # pragma: no cover + raise ImportError( + "DTLS transport requires pyOpenSSL. Install with: pip install goodwe[dtls]" + ) from exc + + +class DTLSInverterProtocol(InverterProtocol): + """Modbus-over-DTLS transport for newer GoodWe Kit-20 firmware. + + Drop-in replacement for :class:`UdpInverterProtocol` when the dongle + speaks DTLS on UDP/8899. Same Modbus RTU framing, same register layout — + only the transport is encrypted. + """ + + #: OpenSSL cipher selection. ``@SECLEVEL=0`` accepts the dongle's + #: self-signed cert; the rest is the OpenSSL default list. Subclass and + #: override this attribute if a different firmware revision needs a + #: different cipher set. + DEFAULT_CIPHER_LIST = b"DEFAULT:@SECLEVEL=0" + + def __init__(self, host: str, port: int, comm_addr: int, + timeout: int = 2, retries: int = 3): + super().__init__(host, port, comm_addr, timeout, retries) + # The DTLS handshake is expensive (~1 s); reuse the session. + self.keep_alive = True + self._sock: Optional[socket.socket] = None + self._conn: Optional["SSL.Connection"] = None + + # ----- command factories (identical to UDP/RTU path) ----- + + def read_command(self, offset: int, count: int) -> ProtocolCommand: + return ModbusRtuReadCommand(self._comm_addr, offset, count) + + def write_command(self, register: int, value: int) -> ProtocolCommand: + return ModbusRtuWriteCommand(self._comm_addr, register, value) + + def write_multi_command(self, offset: int, values: bytes) -> ProtocolCommand: + return ModbusRtuWriteMultiCommand(self._comm_addr, offset, values) + + # ----- async I/O ----- + + async def send_request(self, command: ProtocolCommand) -> Future: + """Send `command` over DTLS, return a Future with the response. + + Retries up to ``self.retries`` times. Behaviour on retry: + - On validation failure (response decoded but didn't match the + command — typically a stale frame from a previous request that + arrived late), the session is **preserved** and the next attempt + drains stale buffered data before sending again. + - On a genuine I/O / SSL exception the session is torn down before + the next attempt, since it's likely no longer usable. + """ + loop = asyncio.get_running_loop() + await self._ensure_lock().acquire() + try: + self.command = command + payload = command.request_bytes() + for attempt in range(self.retries + 1): + if attempt > 0: + logger.debug("Retry #%d/%d for %s", attempt, self.retries, command) + try: + raw = await loop.run_in_executor( + None, self._send_recv_blocking, payload + ) + except Exception as err: # noqa: BLE001 — broad: anything from socket/SSL + logger.debug("DTLS send/recv attempt %d failed: %s", + attempt + 1, err) + # Real I/O / SSL failure → session may be unusable. + await loop.run_in_executor(None, self._close_session) + continue + + if command.validator(raw): + fut = loop.create_future() + # Caller (ProtocolCommand.execute) wraps raw bytes in + # ProtocolResponse itself — match the UDP path's contract. + fut.set_result(raw) + return fut + + # Validation failed → likely a stale leftover from a previous + # timed-out request. Don't close the session; the drain at the + # start of the next _send_recv_blocking will discard the rest. + logger.debug("Response failed validation (stale), retrying on same session.") + + return self._max_retries_reached() + finally: + if self._lock and self._lock.locked(): + self._lock.release() + if not self.keep_alive: + await loop.run_in_executor(None, self._close_session) + + async def close(self) -> None: + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, self._close_session) + + def _close_transport(self) -> None: + # Sync flavour invoked by base class on errors. + self._close_session() + if self.response_future and not self.response_future.done(): + self.response_future.cancel() + + # ----- sync internals (run inside an executor) ----- + + def _send_recv_blocking(self, payload: bytes) -> bytes: + if self._conn is None or self._sock is None: + self._connect_blocking() + # Drain any leftover bytes from a previous (timed-out) request — without + # this they would surface as the response to the new request. + self._drain_stale() + self._conn.send(payload) + deadline = time.monotonic() + self.timeout + self._flush_outbound(deadline) + + while True: + try: + return self._conn.recv(4096) + except (SSL.WantReadError, SSL.WantWriteError): + if time.monotonic() >= deadline: + raise RequestFailedException( + f"No DTLS response within {self.timeout}s", 0 + ) + self._pump(deadline) + + def _connect_blocking(self) -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.connect((self._host, self._port)) + sock.setblocking(False) + + ctx = SSL.Context(SSL.DTLS_METHOD) + ctx.set_verify(SSL.VERIFY_NONE, lambda *_: True) + ctx.set_cipher_list(self.DEFAULT_CIPHER_LIST) + + conn = SSL.Connection(ctx, None) # memory BIOs + conn.set_connect_state() + + # Handshake budget = 5x the per-request timeout, capped at >= 5s. + deadline = time.monotonic() + max(self.timeout * 5, 5.0) + self._sock = sock + self._conn = conn + while True: + try: + conn.do_handshake() + break + except (SSL.WantReadError, SSL.WantWriteError): + self._pump(deadline) + if time.monotonic() >= deadline: + self._close_session() + # The dongle accepts only one concurrent DTLS session in our + # observation, so a hung handshake is most often another + # client (e.g. the vendor's mobile app) holding the slot. + raise RequestFailedException( + f"DTLS handshake to {self._host}:{self._port} timed out " + "(dongle may already have a client connected)", 0 + ) + + logger.debug("DTLS connected to %s:%d cipher=%s proto=%s", + self._host, self._port, + conn.get_cipher_name(), + conn.get_protocol_version_name()) + + def _close_session(self) -> None: + conn, self._conn = self._conn, None + sock, self._sock = self._sock, None + if conn is not None: + try: + conn.shutdown() + except Exception: # noqa: BLE001 + pass + if sock is not None: + try: + sock.close() + except Exception: # noqa: BLE001 + pass + + def _drain_stale(self) -> None: + """Pull any pending bytes off the socket+BIO and discard them.""" + try: + while True: + data = self._sock.recv(4096) + if not data: + break + self._conn.bio_write(data) + except (BlockingIOError, OSError): + pass + while True: + try: + stale = self._conn.recv(4096) + if not stale: + break + except (SSL.WantReadError, SSL.WantWriteError): + break + + def _flush_outbound(self, deadline: float) -> None: + while True: + try: + out = self._conn.bio_read(4096) + except SSL.WantReadError: + return + if not out: + return + self._sock.send(out) + if time.monotonic() >= deadline: + return + + def _pump(self, deadline: float) -> None: + self._flush_outbound(deadline) + wait = max(0.0, min(0.5, deadline - time.monotonic())) + rlist, _, _ = select.select([self._sock], [], [], wait) + if rlist: + data = self._sock.recv(4096) + if data: + self._conn.bio_write(data) diff --git a/goodwe/es.py b/goodwe/es.py index b3befa1..edc0f45 100644 --- a/goodwe/es.py +++ b/goodwe/es.py @@ -242,8 +242,10 @@ def __init__( comm_addr: int = 0, timeout: int = 1, retries: int = 3, + dtls: bool = False, ): - super().__init__(host, port, comm_addr if comm_addr else 0xF7, timeout, retries) + super().__init__(host, port, comm_addr if comm_addr else 0xF7, + timeout, retries, dtls=dtls) self._settings: dict[str, Sensor] = {s.id_: s for s in self.__all_settings} def _supports_eco_mode_v2(self) -> bool: diff --git a/goodwe/et.py b/goodwe/et.py index 8fd84c0..5071e5e 100644 --- a/goodwe/et.py +++ b/goodwe/et.py @@ -695,8 +695,10 @@ def __init__( comm_addr: int = 0, timeout: int = 1, retries: int = 3, + dtls: bool = False, ): - super().__init__(host, port, comm_addr if comm_addr else 0xF7, timeout, retries) + super().__init__(host, port, comm_addr if comm_addr else 0xF7, + timeout, retries, dtls=dtls) self._READ_DEVICE_VERSION_INFO: ProtocolCommand = self._read_command( 0x88B8, 0x0021 ) diff --git a/goodwe/inverter.py b/goodwe/inverter.py index 2892e02..1b2472f 100644 --- a/goodwe/inverter.py +++ b/goodwe/inverter.py @@ -232,9 +232,10 @@ def __init__( comm_addr: int = 0, timeout: int = 1, retries: int = 3, + dtls: bool = False, ): self._protocol: InverterProtocol = self._create_protocol( - host, port, comm_addr, timeout, retries + host, port, comm_addr, timeout, retries, dtls=dtls ) self._consecutive_failures_count: int = 0 @@ -454,8 +455,14 @@ def settings(self) -> tuple[Sensor, ...]: @staticmethod def _create_protocol( - host: str, port: int, comm_addr: int, timeout: int, retries: int + host: str, port: int, comm_addr: int, timeout: int, retries: int, + dtls: bool = False, ) -> InverterProtocol: + if dtls: + # Imported lazily so the optional pyOpenSSL dependency is only + # required when the DTLS path is actually used. + from .dtls import DTLSInverterProtocol # noqa: PLC0415 + return DTLSInverterProtocol(host, port, comm_addr, timeout, retries) if port == GOODWE_UDP_PORT: return UdpInverterProtocol(host, port, comm_addr, timeout, retries) return TcpInverterProtocol(host, port, comm_addr, timeout, retries) diff --git a/setup.cfg b/setup.cfg index 64ad966..ea92255 100644 --- a/setup.cfg +++ b/setup.cfg @@ -28,6 +28,9 @@ python_requires = >= 3.8 [options.packages.find] exclude = tests* +[options.extras_require] +dtls = pyOpenSSL>=24.0 + [flake8] exclude = venv max-complexity = 10 diff --git a/tests/manual_dtls_check.py b/tests/manual_dtls_check.py new file mode 100644 index 0000000..619612a --- /dev/null +++ b/tests/manual_dtls_check.py @@ -0,0 +1,48 @@ +"""Manual integration check for DTLS transport — requires a real dongle. + +Not run by the standard test suite. Invoke directly when verifying against +hardware that advertises ``dtls_port:8899`` in its UDP/48899 discovery response. + +Usage: + python tests/manual_dtls_check.py [family] + +family defaults to DT (single-phase MS / D-NS / XS / PSC etc.). +""" + +import asyncio +import os +import sys + +import goodwe + + +async def main(host: str, family: str) -> int: + print(f"[*] connecting to {host} via DTLS, family={family} ...") + inv = await goodwe.connect(host, family=family, dtls=True, retries=2, timeout=2) + print(f" model: {inv.model_name}") + print(f" serial: {inv.serial_number}") + print(f" rated: {inv.rated_power} W") + + data = await inv.read_runtime_data() + print() + print("Sample sensors:") + keys = ("timestamp", "vgrid1", "fgrid1", "igrid1", "total_power", + "temperature", "e_day", "e_total", "h_total", "work_mode") + for k in keys: + v = data.get(k) + unit = next((s.unit for s in inv.sensors() if s.id_ == k), "") + print(f" {k:>15s} = {v} {unit}") + + await inv._protocol.close() + print() + print("[+] DTLS transport round-trip OK") + return 0 + + +if __name__ == "__main__": + if len(sys.argv) < 2: + print(__doc__) + sys.exit(2) + host = sys.argv[1] + family = sys.argv[2] if len(sys.argv) > 2 else "DT" + sys.exit(asyncio.run(main(host, family))) diff --git a/tests/test_dtls.py b/tests/test_dtls.py new file mode 100644 index 0000000..ed0978e --- /dev/null +++ b/tests/test_dtls.py @@ -0,0 +1,169 @@ +"""Tests for the DTLS transport. + +The DTLS layer itself (pyOpenSSL) is not re-tested here — these tests verify +the async wrapper, retry/lock semantics, and command-construction symmetry with +the legacy UDP path. The blocking ``_send_recv_blocking`` and ``_close_session`` +methods are patched out; pyOpenSSL is exercised end-to-end against real hardware +in ``tests/manual_dtls_check.py`` (excluded from the standard test run). +""" + +import asyncio +from unittest import IsolatedAsyncioTestCase, TestCase +from unittest.mock import MagicMock, patch + +from goodwe.dtls import DTLSInverterProtocol +from goodwe.exceptions import RequestFailedException +from goodwe.protocol import ( + ModbusRtuReadCommand, + ModbusRtuWriteCommand, + ModbusRtuWriteMultiCommand, +) + + +# Canned bytes representing the dongle's response to FC03 read of 1 register at 0x7594. +# Format: AA 55 [slave] [fc] [byte_count] [hi lo] [crc16_lo crc16_hi] +_VALID_FC03_RESPONSE = bytes.fromhex("aa55f70302" + "1A04" + "") # placeholder, fixed below + + +def _build_fc03_response(comm_addr: int, register_count: int, data: bytes) -> bytes: + """Build a wrapped Modbus RTU response that the existing validator accepts.""" + from goodwe.modbus import _modbus_checksum, MODBUS_READ_CMD + body = bytearray() + body.append(0xAA) # AA55 wrapper byte 0 + body.append(0x55) # AA55 wrapper byte 1 + body.append(comm_addr) + body.append(MODBUS_READ_CMD) + body.append(register_count * 2) # byte count + body.extend(data) + crc = _modbus_checksum(bytes(body[2:])) # CRC over slave..data + body.append(crc & 0xFF) + body.append((crc >> 8) & 0xFF) + return bytes(body) + + +class TestDTLSCommandConstruction(TestCase): + """The DTLS protocol class must produce the same Modbus RTU framing as + UdpInverterProtocol — only the transport differs.""" + + def setUp(self): + self.proto = DTLSInverterProtocol("192.0.2.1", 8899, comm_addr=0xF7) + + def test_read_command_returns_modbus_rtu(self): + cmd = self.proto.read_command(0x7594, 0x49) + self.assertIsInstance(cmd, ModbusRtuReadCommand) + + def test_write_command_returns_modbus_rtu(self): + cmd = self.proto.write_command(40328, 50) + self.assertIsInstance(cmd, ModbusRtuWriteCommand) + + def test_write_multi_command_returns_modbus_rtu(self): + cmd = self.proto.write_multi_command(40310, b"\x01\x02\x03\x04") + self.assertIsInstance(cmd, ModbusRtuWriteMultiCommand) + + def test_keep_alive_default_is_true(self): + # Re-handshaking per request would cost ~1s — DTLS sessions must persist. + self.assertTrue(self.proto.keep_alive) + + +class TestDTLSSendRequest(IsolatedAsyncioTestCase): + """Verify async send_request — uses the executor + retry + lock plumbing, + with the actual DTLS I/O patched out.""" + + def setUp(self): + self.proto = DTLSInverterProtocol( + "192.0.2.1", 8899, comm_addr=0xF7, timeout=1, retries=2 + ) + + async def test_successful_round_trip(self): + # Read 1 register at 0x7594, dongle responds with one big-endian U16 = 0x1A04. + cmd = self.proto.read_command(0x7594, 1) + canned = _build_fc03_response(0xF7, 1, b"\x1A\x04") + + with patch.object(self.proto, "_send_recv_blocking", + return_value=canned) as mocked: + future = await self.proto.send_request(cmd) + self.assertEqual(future.result(), canned) + mocked.assert_called_once() + # The dispatched payload is the request bytes built by ModbusRtuReadCommand. + (sent_payload,), _ = mocked.call_args + self.assertEqual(sent_payload[0], 0xF7) # slave addr + self.assertEqual(sent_payload[1], 0x03) # function code FC03 + + async def test_retry_on_transient_failure(self): + cmd = self.proto.read_command(0x7594, 1) + canned = _build_fc03_response(0xF7, 1, b"\x12\x34") + + attempts = [] + + def flaky_send(payload): + attempts.append(payload) + if len(attempts) == 1: + raise RequestFailedException("simulated UDP loss", 0) + return canned + + with patch.object(self.proto, "_send_recv_blocking", side_effect=flaky_send): + with patch.object(self.proto, "_close_session"): # don't actually touch sockets + future = await self.proto.send_request(cmd) + self.assertEqual(future.result(), canned) + self.assertEqual(len(attempts), 2) # one fail, one success + + async def test_max_retries_exhausted(self): + cmd = self.proto.read_command(0x7594, 1) + + with patch.object( + self.proto, "_send_recv_blocking", + side_effect=RequestFailedException("persistent failure", 0), + ): + with patch.object(self.proto, "_close_session"): + future = await self.proto.send_request(cmd) + with self.assertRaises(Exception): + future.result() + + async def test_stale_response_triggers_retry(self): + """If the dongle returns a wrong-length response (stale from a prior + request), the validator rejects it and we retry on the next attempt.""" + cmd = self.proto.read_command(0x7594, 1) # asks for 1 register => 2 bytes + stale = _build_fc03_response(0xF7, 2, b"\x00\x00\x00\x00") + good = _build_fc03_response(0xF7, 1, b"\x12\x34") + + send_calls = [] + + def respond(payload): + send_calls.append(payload) + return stale if len(send_calls) == 1 else good + + with patch.object(self.proto, "_send_recv_blocking", side_effect=respond): + with patch.object(self.proto, "_close_session"): + future = await self.proto.send_request(cmd) + self.assertEqual(future.result(), good) + # The same payload was sent twice — session continued, retry happened. + self.assertEqual(len(send_calls), 2) + self.assertEqual(send_calls[0], send_calls[1]) + + async def test_io_exception_recovers_via_retry(self): + """On a genuine I/O failure (not just a stale response), the next + attempt still succeeds — the retry/lock plumbing handles both cases.""" + cmd = self.proto.read_command(0x7594, 1) + good = _build_fc03_response(0xF7, 1, b"\x12\x34") + + send_calls = [] + + def flaky(payload): + send_calls.append(payload) + if len(send_calls) == 1: + raise OSError("simulated socket failure") + return good + + with patch.object(self.proto, "_send_recv_blocking", side_effect=flaky): + with patch.object(self.proto, "_close_session"): + future = await self.proto.send_request(cmd) + self.assertEqual(future.result(), good) + self.assertEqual(len(send_calls), 2) + + +class TestDTLSCloseLifecycle(IsolatedAsyncioTestCase): + async def test_close_calls_close_session(self): + proto = DTLSInverterProtocol("192.0.2.1", 8899, comm_addr=0xF7) + with patch.object(proto, "_close_session") as mocked: + await proto.close() + mocked.assert_called_once()