From d67db0a4be44c1684f663e9f5417c1d9c80b32b1 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sun, 6 Sep 2026 01:09:01 +0000 Subject: [PATCH] Fix the issues found in the review of 1.0.3 A fourth review, of 1.0.3, drove the real socket path that the test suite fakes and found two things the original library did that this one had lost: a socket that goes bad is replaced, and a hostname that does not resolve fails at once. This fixes both, carries one device fix from upstream, and takes the small items along. Tested on 3.13 and 3.14, 269 tests under -X dev with ResourceWarning as an error, and live against an RM4 Pro including a forced socket failure and a bad hostname. Technical details: - _Protocol.error_received and connection_lost put the error on the receive queue, so a waiting request fails at once with that OSError (send failures: no route, address gone) instead of waiting out the timeout. ICMP port unreachable is logged and treated as silence, which is what the original's unconnected socket saw, so the timeout decides as before. - A request that fails for a network reason drops the endpoint; the next call opens a fresh socket, as the original did by opening one per call. - scan(), ping() and send_setup_packet() resolve the destination once with loop.getaddrinfo and let socket.gaierror propagate; a send failure in ping() and setup() is raised. Nothing resolves on the loop any more. - a2._send uses the SP4/LB1 frame layout (length 12, four-byte data length). The bytes match upstream #826, tested by its author on a real A2; the one oracle case re-recorded on purpose and noted in harness.py. - xdiscover() closes the scan() generator it wraps; captured_at is excluded from CapturedSignal equality; async generator functions are annotated AsyncGenerator (mypy 40 to 37); TICK docstring matches the README. - tests/test_loopback.py drives the real datagram endpoint, including 20 concurrent requests and the ICMP path on loopback. - README: errors discover()/hello() raise, the socket drop on failure, where CaptureInProgressError can come from. - Version 1.0.4. Co-authored-by: kanshurichard <29994770+kanshurichard@users.noreply.github.com> --- CHANGELOG.md | 53 +++++++++++++ README.md | 20 ++++- broadlink/__init__.py | 12 +-- broadlink/device.py | 102 +++++++++++++++++++++---- broadlink/remote.py | 19 ++--- broadlink/sensor.py | 54 ++++++------- pyproject.toml | 2 +- tests/oracle/fixtures.json | 2 +- tests/oracle/harness.py | 7 ++ tests/test_loopback.py | 110 +++++++++++++++++++++++++++ tests/test_transport.py | 150 +++++++++++++++++++++++++++++++++++-- 11 files changed, 461 insertions(+), 70 deletions(-) create mode 100644 tests/test_loopback.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2413a7a3..bbc9c4cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,59 @@ All notable changes to this project are recorded here. The format follows Keep a Changelog; versions follow Semantic Versioning. +## 1.0.4 - 2026-09-06 + +Fixes from a fourth review, of 1.0.3, which drove the real socket path the +test suite fakes and found two behaviours the original library had and +this one had lost. One device fix carried from upstream. + +### Fixed + +- A connected socket that went bad (interface bounce, host address change, + container network restart) was never replaced: the request waited out + its timeout and every later request did the same until `aclose()`. The + original opened a socket per call, so it healed on the next one. Now a + send failure the socket reports (no route, address gone) fails the + waiting request at once with that `OSError`, and a request that fails + for a network reason (that, or a timeout) drops the socket so the next + call opens a fresh one. A transport asyncio closes from its side also + wakes the waiting request instead of leaving it to time out. An ICMP + "port unreachable" (a host that is up with nothing listening, or a + device mid-reboot) is logged and treated as silence, since the + original's unconnected socket never saw those, so the timeout decides + as before. +- `discover()`, `hello()`, `ping()` and `setup()` passed hostnames straight + to `sendto`, which resolved them with a blocking call on the event loop + and swallowed the failure: a name that did not resolve made `hello()` + wait out its timeout and `ping()` return without sending. The + destination is now resolved once, off the loop, and `socket.gaierror` + propagates as it did from the original's socket. A send failure in + `ping()` and `setup()` is raised too. +- The A2 air quality sensor's request frame was two bytes short and + declared the wrong length, and real units answered every read with + error -5. The frame now follows the SP4/LB1 layout, which is byte for + byte the packet upstream pull request #826 tested on an A2. That is the + one oracle case re-recorded on purpose; the fix is carried on the + strength of that report, not of hardware we have. +- `xdiscover()` closes the `scan()` generator it wraps, so the discovery + socket is closed when the caller stops iterating rather than by the + finalizer a few turns later (1.0.2 claimed this and only `Device.hello()` + did it). +- Two identical captures compare equal: `CapturedSignal.captured_at` no + longer takes part in equality or hashing. +- Async generator functions are annotated `AsyncGenerator`, which has the + `aclose()` the library and the README call; `AsyncIterator` does not. +- The `TICK` docstring tells the same story as the README: 8192/269 from + protocol.md's measured conversion, not a 32768 Hz clock. + +### Added + +- A loopback test module that drives the real datagram endpoint, including + the socket-error path, since every other transport test fakes it. +- README: which errors `discover()` and `hello()` raise, that a failed + request drops its socket, and that `CaptureInProgressError` can come + from the `capture()` call or from the first iteration. + ## 1.0.3 - 2026-09-06 Fixes from a third review, this one of 1.0.2. No change to the wire diff --git a/README.md b/README.md index e8faf583..329348f2 100644 --- a/README.md +++ b/README.md @@ -162,6 +162,11 @@ If the device is locked, it may not be discoverable with broadcast. In such case device = await broadlink.hello("192.168.0.16") ``` +`discover()` and `hello()` raise `NetworkTimeoutError` when nothing answers +within the timeout, `socket.gaierror` when a hostname does not resolve, and +`OSError` when the socket cannot be opened or the send fails (no route, for +example), the same errors the original library raised from its socket. + If you are a perfomance freak, use `broadlink.xdiscover()` to create devices instantly: ```python3 async for device in broadlink.xdiscover(): @@ -202,9 +207,14 @@ await device.aclose() The socket reopens by itself on the next call, so closing is cheap and safe to do at any time. A request that is in flight when `aclose()` runs -fails with `EndpointClosedError`. An integration that creates devices -should close them when it unloads; a device that is never closed holds -its socket until it is garbage collected. +fails with `EndpointClosedError`. A request that fails for a network +reason (a timeout, or an `OSError` from the socket such as "network is +unreachable" after an interface change, raised at once) also drops the socket, so the +next call starts fresh rather than reusing one that has gone bad, which +is how the original library behaved by opening a socket per call. An +integration that creates devices should close them when it unloads; a +device that is never closed holds its socket until it is garbage +collected. The next steps depend on the type of device you want to control. @@ -280,7 +290,9 @@ By default the window closes after the first signal. Pass because the device holds only one code per learning session. A universal remote has a single receiver, so only one capture window can be open on a device at a time: opening a second one raises `CaptureInProgressError` -while the first is still held. Always close a window you leave early +while the first is still held, either from the `capture()` call itself or +from the new window's first iteration, depending on what the first window +was doing at that moment. Always close a window you leave early (`aclosing` above does it), otherwise it stays open until Python collects the generator. diff --git a/broadlink/__init__.py b/broadlink/__init__.py index f68df1d6..0dd588d3 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -1,7 +1,7 @@ """The python-broadlink library.""" import contextlib -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from . import exceptions as e from .alarm import S1C @@ -294,15 +294,15 @@ async def xdiscover( local_ip_address: str | None = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, -) -> AsyncIterator[Device]: +) -> AsyncGenerator[Device]: """Discover devices connected to the local network. Yields each device as soon as it answers. """ - async for resp in scan( - timeout, local_ip_address, discover_ip_address, discover_ip_port - ): - yield gendevice(*resp) + responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port) + async with contextlib.aclosing(responses): + async for resp in responses: + yield gendevice(*resp) # Setup a new Broadlink device via AP Mode. Review the README to see how to enter AP Mode. diff --git a/broadlink/device.py b/broadlink/device.py index 600356ea..c28f3006 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -15,7 +15,7 @@ import logging import random import socket -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes @@ -48,12 +48,20 @@ _CLOSED = (None, None) """Sentinel put on the receive queue when the endpoint is closed.""" +_QueueItem = tuple[bytes | Exception | None, tuple[str, int] | None] +"""What the receive queue carries: a datagram with its source address, an +error the socket reported (address ``None``), or ``_CLOSED``.""" + class _Protocol(asyncio.DatagramProtocol): - """Datagram protocol that hands every received packet to a queue.""" + """Datagram protocol that hands every received packet to a queue. + + Errors the socket reports go on the same queue, so the request that is + waiting fails at once instead of waiting out its timeout. + """ def __init__(self) -> None: - self.queue: asyncio.Queue[tuple[bytes, tuple[str, int]]] = asyncio.Queue() + self.queue: asyncio.Queue[_QueueItem] = asyncio.Queue() self.transport: asyncio.DatagramTransport | None = None def connection_made(self, transport: asyncio.BaseTransport) -> None: @@ -65,18 +73,30 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: self.queue.put_nowait((data, addr)) def error_received(self, exc: Exception) -> None: - # ICMP unreachable and the like. Surface it as a receive of nothing; - # the retry loop will time out and raise NetworkTimeoutError. - pass + """Queue a send failure or an ICMP error for the waiting request.""" + self.queue.put_nowait((exc, None)) def connection_lost(self, exc: Exception | None) -> None: - """Nothing to do; a waiting request is told through the queue.""" + """Wake the waiting request if asyncio closed the transport on us.""" + self.queue.put_nowait((exc, None) if exc is not None else _CLOSED) def drain(self) -> None: """Drop anything that arrived before the current request.""" while not self.queue.empty(): self.queue.get_nowait() + def raise_if_error(self) -> None: + """Raise the error a send just reported, if it reported one. + + asyncio delivers a failed ``sendto`` to ``error_received`` before + ``sendto`` returns, so a fire-and-forget sender can check right + after sending and raise the ``OSError`` the way a plain socket did. + """ + while not self.queue.empty(): + item, _ = self.queue.get_nowait() + if isinstance(item, Exception): + raise item + async def _open_endpoint( local_addr: tuple[str, int] | None = None, @@ -95,6 +115,21 @@ async def _open_endpoint( return transport, protocol # type: ignore[return-value] +async def _resolve(host: str, port: int) -> tuple[str, int]: + """Resolve a destination once, off the event loop. + + Sending to a hostname through an unconnected datagram socket would + resolve it with a blocking call on the loop and hide the failure. A + name that does not resolve raises ``socket.gaierror`` here, as the + original library's ``sendto`` did. + """ + loop = asyncio.get_running_loop() + info = await loop.getaddrinfo( + host, port, family=socket.AF_INET, type=socket.SOCK_DGRAM + ) + return info[0][4][:2] # type: ignore[return-value] + + def _hello_packet(local_ip_address: str, port: int) -> bytearray: packet = bytearray(0x30) packet[0x08:0x14] = Datetime.pack(Datetime.now()) @@ -119,13 +154,14 @@ async def scan( local_ip_address: str | None = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, -) -> AsyncIterator[HelloResponse]: +) -> AsyncGenerator[HelloResponse]: """Broadcast a hello message and yield responses as they arrive. The hello is repeated every ``DEFAULT_RETRY_INTVL`` seconds until ``timeout`` elapses. Each device is yielded once. """ local_addr = (local_ip_address, 0) if local_ip_address else None + target = await _resolve(discover_ip_address, discover_ip_port) transport, protocol = await _open_endpoint(local_addr=local_addr, broadcast=True) try: if local_ip_address: @@ -140,7 +176,7 @@ async def scan( discovered: set[tuple[tuple[str, int], bytes, int]] = set() while (loop.time() - start) < timeout: - transport.sendto(packet, (discover_ip_address, discover_ip_port)) + transport.sendto(packet, target) deadline = min(DEFAULT_RETRY_INTVL, timeout - (loop.time() - start)) slot_end = loop.time() + deadline while True: @@ -151,7 +187,11 @@ async def scan( resp, host = await asyncio.wait_for(protocol.queue.get(), remaining) except TimeoutError: break - if len(resp) < 0x80: + if resp is None: + return # The transport was closed under us. + if isinstance(resp, Exception): + raise resp + if host is None or len(resp) < 0x80: continue entry = _parse_hello(resp, host) key = (entry[1], entry[2], entry[0]) @@ -167,9 +207,11 @@ async def send_setup_packet( payload: bytes, ip_address: str, port: int = DEFAULT_PORT ) -> None: """Broadcast one Wi-Fi provisioning packet to a device in AP mode.""" - transport, _ = await _open_endpoint(broadcast=True) + target = await _resolve(ip_address, port) + transport, protocol = await _open_endpoint(broadcast=True) try: - transport.sendto(payload, (ip_address, port)) + transport.sendto(payload, target) + protocol.raise_if_error() finally: transport.close() @@ -181,11 +223,13 @@ async def ping(ip_address: str, port: int = DEFAULT_PORT) -> None: Useful to prevent reboots when the cloud cannot be reached. It must be sent every 2 minutes in such cases. """ - transport, _ = await _open_endpoint(broadcast=True) + target = await _resolve(ip_address, port) + transport, protocol = await _open_endpoint(broadcast=True) try: packet = bytearray(0x30) packet[0x26] = 1 - transport.sendto(packet, (ip_address, port)) + transport.sendto(packet, target) + protocol.raise_if_error() finally: transport.close() @@ -407,6 +451,22 @@ async def aclose(self) -> None: if protocol is not None: protocol.queue.put_nowait(_CLOSED) # type: ignore[arg-type] + def _drop_endpoint(self) -> None: + """Throw the endpoint away after a failure; the next call reopens it. + + A connected datagram socket can go bad for good (the interface + bounced, the host's address changed), and the original library + never noticed because it opened a socket per call. Dropping the + endpoint whenever a request fails restores that self-healing. + """ + transport = self._transport + self._transport = None + self._protocol = None + self._endpoint_addr = None + if transport is not None: + transport.close() + _LOGGER.debug("%s: endpoint dropped after a failure", self.host[0]) + async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: if self._transport is not None and self._endpoint_addr != self.host: # The caller changed host; the connected socket points at the @@ -510,6 +570,19 @@ async def _exchange(self, packet: bytes) -> bytes: raise e.EndpointClosedError( -4013, "Endpoint closed", "The device endpoint was closed" ) + if isinstance(resp, ConnectionRefusedError): + # ICMP port unreachable: the host is up and nothing is + # listening, or the device is rebooting. The original + # library's unconnected socket never saw these, so keep + # waiting and let the timeout decide, as it did. + _LOGGER.debug("%s: port unreachable, still waiting", self.host[0]) + continue + if isinstance(resp, Exception): + # A send failure (no route, address gone) or a fatal + # transport error: fail now and throw the socket away. + _LOGGER.debug("%s: socket error: %s", self.host[0], resp) + self._drop_endpoint() + raise resp resp = self._validate(resp) reply_count = int.from_bytes(resp[0x28:0x2A], "little") if reply_count == count or reply_count not in self._recent: @@ -521,6 +594,7 @@ async def _exchange(self, packet: bytes) -> bytes: ) if loop.time() - start >= timeout: _LOGGER.debug("%s: no reply within %ss", self.host[0], timeout) + self._drop_endpoint() raise e.NetworkTimeoutError( -4000, "Network timeout", diff --git a/broadlink/remote.py b/broadlink/remote.py index 5fad6d1a..6840833c 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -6,7 +6,7 @@ import struct import time import weakref -from collections.abc import AsyncIterator, Awaitable, Callable +from collections.abc import AsyncGenerator, Awaitable, Callable from dataclasses import dataclass, field from typing import Self @@ -18,9 +18,10 @@ TICK = 8192 / 269 """Duration of one Broadlink timing unit in microseconds (about 30.45 us). -The RM firmware counts pulses on a 32768 Hz clock (protocol.md: us * 269 / 8192). -Earlier releases used 32.84, the inverse of the right ratio applied the wrong -way round, which compressed externally sourced IR codes by about 7 percent +The value comes from protocol.md, whose conversion "us * 269 / 8192 works +very well" was measured against real firmware; 8192/269 is its inverse. +Earlier releases used 32.84, the right ratio applied the wrong way round, +which compressed externally sourced IR codes by about 7 percent (mjg59/python-broadlink#839). Codes learned and replayed through the same device were unaffected because both directions shared the constant. """ @@ -176,7 +177,7 @@ class CapturedSignal: repeat: int = 0 frequency_mhz: float | None = None type_byte: int | None = None - captured_at: float = field(default_factory=time.time, repr=False) + captured_at: float = field(default_factory=time.time, repr=False, compare=False) @classmethod def from_packet( @@ -300,7 +301,7 @@ def capture( stop_after_first: bool = True, poll_interval: float = DEFAULT_POLL_INTERVAL, rearm_interval: float = DEFAULT_REARM_INTERVAL, - ) -> AsyncIterator[CapturedSignal]: + ) -> AsyncGenerator[CapturedSignal]: """Open an infrared capture window and yield what the device hears. The device is put into learning mode and polled every @@ -348,7 +349,7 @@ async def _capture_loop( frequency_mhz: float | None, *, claim: list | None = None, - ) -> AsyncIterator[CapturedSignal]: + ) -> AsyncGenerator[CapturedSignal]: # ``claim`` carries a weak reference to this generator (filled in by # the caller after creating it); None means the caller owns the # window claim, as capture_rf does for its inner loop. @@ -457,7 +458,7 @@ def capture_rf( stop_after_first: bool = True, poll_interval: float = DEFAULT_POLL_INTERVAL, rearm_interval: float = DEFAULT_REARM_INTERVAL, - ) -> AsyncIterator[CapturedSignal]: + ) -> AsyncGenerator[CapturedSignal]: """Open a radio frequency capture window and yield what the device hears. With ``frequency`` (in MHz, for example 433.92) the device goes @@ -496,7 +497,7 @@ async def _capture_rf_loop( rearm_interval: float, *, claim: list, - ) -> AsyncIterator[CapturedSignal]: + ) -> AsyncGenerator[CapturedSignal]: if window < 0 or poll_interval <= 0: raise ValueError("window must be 0 or positive, poll_interval positive") await self._claim_window(claim[0]) diff --git a/broadlink/sensor.py b/broadlink/sensor.py index 3091707d..a0f86abe 100644 --- a/broadlink/sensor.py +++ b/broadlink/sensor.py @@ -1,6 +1,6 @@ """Support for sensors.""" -from collections.abc import Sequence +import struct from . import exceptions as e from .device import Device @@ -48,35 +48,35 @@ class a2(Device): TYPE = "A2" - async def _send(self, operation: int, data: Sequence = b""): - """Send a command to the device.""" - packet = bytearray(12) - packet[0x02] = 0xA5 - packet[0x03] = 0xA5 - packet[0x04] = 0x5A - packet[0x05] = 0x5A - packet[0x08] = operation - packet[0x09] = 0x0B - - if data: - data_len = len(data) - packet[0x0A] = data_len & 0xFF - packet[0x0B] = data_len >> 8 - packet += bytes(2) - packet.extend(data) - - checksum = sum(packet, 0xBEAF) & 0xFFFF - packet[0x06] = checksum & 0xFF - packet[0x07] = checksum >> 8 - - packet_len = len(packet) - 2 - packet[0x00] = packet_len & 0xFF - packet[0x01] = packet_len >> 8 + async def _send(self, operation: int, data: bytes = b"") -> bytes: + """Send a command to the device. + + The frame is the one the SP4 and LB1 families use: a two-byte + length, the A5A5 5A5A marker, a checksum, the operation, 0x0B and + a four-byte data length. The 0.19.0 code wrote a two-byte data + length and a length field two short, and real A2 units answered + every request with error -5 (mjg59/python-broadlink#826). + """ + packet = bytearray(14) + struct.pack_into( + " dict: """Return the state of the sensors in raw format.""" diff --git a/pyproject.toml b/pyproject.toml index af17ec80..ffa751f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.3" +version = "1.0.4" description = "Python API for controlling Broadlink devices" readme = "README.md" license = "MIT" diff --git a/tests/oracle/fixtures.json b/tests/oracle/fixtures.json index 782b2bae..170db6ef 100644 --- a/tests/oracle/fixtures.json +++ b/tests/oracle/fixtures.json @@ -2688,7 +2688,7 @@ "sent": [ [ 106, - "0a00a5a55a5ab9c0010b0000" + "0c00a5a55a5ab9c0010b00000000" ] ], "unused_responses": 0 diff --git a/tests/oracle/harness.py b/tests/oracle/harness.py index 8c5c5254..3beb0c84 100644 --- a/tests/oracle/harness.py +++ b/tests/oracle/harness.py @@ -18,6 +18,13 @@ The runner accepts awaitables so the same cases can drive an asynchronous ``send_packet`` later without changing the cases. + +Deliberate departures from 0.19.0, re-recorded on purpose and reviewed in +the pull request that made them: + +- ``a2.check_sensors_raw`` (1.0.4): the request frame follows the SP4/LB1 + layout (length 12, four-byte data length) instead of the 0.19.0 frame + the device rejected with error -5. Upstream #826. """ from __future__ import annotations diff --git a/tests/test_loopback.py b/tests/test_loopback.py new file mode 100644 index 00000000..4921fcaf --- /dev/null +++ b/tests/test_loopback.py @@ -0,0 +1,110 @@ +"""The real datagram path, on loopback. + +Every other transport test replaces ``_open_endpoint`` with a fake, so the +lines that talk to asyncio's real datagram transport (``_Protocol``, +``_open_endpoint``) are only exercised here. A small fake device answers on +127.0.0.1 with real sockets. +""" + +from __future__ import annotations + +import asyncio +import socket +import sys + +import pytest + +from broadlink import device as device_module +from broadlink import exceptions as e +from broadlink.device import Device +from tests.oracle.harness import MAC, make_response + + +class FakeDevice(asyncio.DatagramProtocol): + """Answers every request frame with a canned payload, counter echoed.""" + + def __init__(self, dev: Device, payload: bytes) -> None: + self.dev = dev + self.payload = payload + self.received: list[bytes] = [] + self.transport: asyncio.DatagramTransport | None = None + + def connection_made(self, transport) -> None: + self.transport = transport + + def datagram_received(self, data: bytes, addr) -> None: + self.received.append(data) + frame = bytearray(make_response(self.dev, self.payload)) + frame[0x28:0x2A] = data[0x28:0x2A] + checksum = sum(frame, 0xBEAF) - sum(frame[0x20:0x22]) & 0xFFFF + frame[0x20:0x22] = checksum.to_bytes(2, "little") + assert self.transport is not None + self.transport.sendto(bytes(frame), addr) + + +async def start_fake(dev: Device, payload: bytes): + loop = asyncio.get_running_loop() + transport, protocol = await loop.create_datagram_endpoint( + lambda: FakeDevice(dev, payload), local_addr=("127.0.0.1", 0) + ) + return transport, protocol, transport.get_extra_info("sockname")[:2] + + +def unused_udp_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def test_request_and_concurrent_requests_over_real_sockets(): + async def go(): + dev = Device(("127.0.0.1", 1), MAC, 0x2737, name="Loopback") + transport, fake, addr = await start_fake(dev, bytes([7]) + bytes(15)) + dev.host = addr + try: + async with dev: + resp = await dev.send_packet(0x6A, b"") + assert dev.decrypt(resp[0x38:])[0] == 7 + results = await asyncio.gather( + *(dev.send_packet(0x6A, bytes([i])) for i in range(20)) + ) + assert all(dev.decrypt(r[0x38:])[0] == 7 for r in results) + assert isinstance(dev._protocol, device_module._Protocol) + assert dev._transport is None + return len(fake.received) + finally: + transport.close() + + assert asyncio.run(go()) == 21 + + +@pytest.mark.skipif(sys.platform == "win32", reason="ICMP errors surface differently") +def test_endpoint_heals_after_a_socket_error_on_loopback(caplog): + """A request to a loopback port nobody listens on draws an ICMP port + unreachable, which the connected socket reports on its next read. That + one is treated like silence (the original's unconnected socket never + saw it) so the request times out as before, but it must have reached + the protocol, and the device must reopen its socket for the next call + instead of reusing the dead one.""" + caplog.set_level("DEBUG", logger="broadlink.device") + + async def go(): + dev = Device(("127.0.0.1", unused_udp_port()), MAC, 0x2737, name="Loopback") + dev.timeout = 0.3 + try: + with pytest.raises(e.NetworkTimeoutError): + await dev.send_packet(0x6A, b"") + assert dev._transport is None + # Point it at a live fake device: the next call opens a new socket. + transport, _, addr = await start_fake(dev, bytes([9]) + bytes(15)) + try: + dev.host = addr + resp = await dev.send_packet(0x6A, b"") + return dev.decrypt(resp[0x38:])[0] + finally: + transport.close() + finally: + await dev.aclose() + + assert asyncio.run(go()) == 9 + assert "port unreachable" in caplog.text diff --git a/tests/test_transport.py b/tests/test_transport.py index 770a45e0..ff5d6370 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -233,22 +233,52 @@ def stamped(dev: Device, payload: bytes, count: int, error: int = 0) -> bytes: return bytes(frame) -def test_late_reply_to_timed_out_request_is_not_taken_as_next_reply(net): - """The defect that 0.19.0 could not have because it threw its socket away - after every call: a slow answer to request 1 arriving after request 1 - timed out must not be returned as the answer to request 2.""" +def test_timed_out_request_drops_its_endpoint(net): + """A request that times out throws its socket away, as 0.19.0 did by + opening one per call, so a socket that has gone bad heals on the next + call and a late reply to the timed-out request lands on a port nobody + listens to any more.""" dev = fixed_device() dev.timeout = 0.02 async def go(): await dev._endpoint() - ep = net.endpoints[-1] + first = net.endpoints[-1] with pytest.raises(e.NetworkTimeoutError): - await dev.send_packet(0x6A, b"") # request 1, count 0x8001, no answer + await dev.send_packet(0x6A, b"") + assert first.closed + assert dev._transport is None + # The late answer arrives on the old socket; the next request opens + # a new one and only sees its own reply. + late = stamped(dev, bytes([1]) + bytes(15), 0x8001) + first.protocol.queue.put_nowait((late, HOST)) + dev.timeout = 1 + net.replies = [(stamped(dev, bytes([2]) + bytes(15), 0x8002), HOST)] + resp = await dev.send_packet(0x6A, b"") + assert net.endpoints[-1] is not first + return dev.decrypt(resp[0x38:])[0] + + assert run(go()) == 2 + + +def test_late_reply_to_cancelled_request_is_not_taken_as_next_reply(net): + """The endpoint survives a cancelled request, so a slow answer to it can + arrive while the next request is waiting on the same socket. It must + not be returned as the answer to that request.""" + dev = fixed_device() + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + task = asyncio.get_running_loop().create_task(dev.send_packet(0x6A, b"")) + await asyncio.sleep(0.005) # request 1 (count 0x8001) is on the wire + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert dev._transport is not None # Its late reply lands while request 2 (count 0x8002) is waiting. late = stamped(dev, bytes([1]) + bytes(15), 0x8001) good = stamped(dev, bytes([2]) + bytes(15), 0x8002) - ep.replies = [late, good] and [] ep.protocol.queue.put_nowait((late, HOST)) async def answer_later(): @@ -256,7 +286,6 @@ async def answer_later(): ep.protocol.queue.put_nowait((good, HOST)) asyncio.get_running_loop().create_task(answer_later()) - dev.timeout = 1 resp = await dev.send_packet(0x6A, b"") return dev.decrypt(resp[0x38:])[0] @@ -424,6 +453,53 @@ async def go(): assert net.endpoints[0].closed +def test_send_failure_fails_the_request_fast_and_heals(net): + """A connected socket whose send fails (route gone, address changed) + reports it through error_received. The request must fail with that + OSError at once, not after the timeout, and the next call must get a + fresh socket rather than the dead one.""" + dev = fixed_device() + dev.timeout = 5 + + async def go(): + await dev._endpoint() + bad = net.endpoints[-1] + + def failing_sendto(data, addr=None): + bad.protocol.error_received(OSError(101, "Network is unreachable")) + + bad.sendto = failing_sendto + t0 = asyncio.get_running_loop().time() + with pytest.raises(OSError) as err: + await dev.send_packet(0x6A, b"") + assert err.value.errno == 101 + assert asyncio.get_running_loop().time() - t0 < 1.0 + assert bad.closed + assert dev._transport is None + net.replies = [(make_response(dev, b""), HOST)] + await dev.send_packet(0x6A, b"") + return net.endpoints[-1] is not bad + + assert run(go()) + + +def test_transport_lost_with_error_wakes_the_request(net): + """If asyncio closes the transport from its side, the waiting request + is told instead of waiting out its timeout.""" + dev = fixed_device() + dev.timeout = 5 + + async def go(): + task = asyncio.get_running_loop().create_task(dev.send_packet(0x6A, b"")) + await asyncio.sleep(0.005) + net.endpoints[-1].protocol.connection_lost(OSError(22, "Invalid argument")) + with pytest.raises(OSError) as err: + await task + return err.value.errno + + assert run(go()) == 22 + + # ------------------------------------------------------------------------- auth @@ -768,6 +844,64 @@ def test_ping_packet(net): assert net.endpoints[-1].closed +def test_unresolvable_hostname_raises_at_once(net, monkeypatch): + """0.19.0 raised socket.gaierror from sendto for a bad hostname. The + async version resolves the name off the loop first and lets the same + error through, instead of waiting out the timeout (hello) or sending + nothing and returning (ping).""" + + async def no_such_host(host, port): + raise socket.gaierror(socket.EAI_NONAME, "Name or service not known") + + monkeypatch.setattr(device_module, "_resolve", no_such_host) + + async def go(): + loop = asyncio.get_running_loop() + t0 = loop.time() + with pytest.raises(socket.gaierror): + await broadlink.hello("nonexistent.invalid", timeout=5) + with pytest.raises(socket.gaierror): + await broadlink.ping("nonexistent.invalid") + with pytest.raises(socket.gaierror): + await broadlink.setup("ssid", "pass", 3, ip_address="nonexistent.invalid") + return loop.time() - t0 + + assert run(go()) < 1.0 + assert net.endpoints == [] # nothing was opened for a name that failed + + +def test_resolve_returns_a_numeric_address(): + assert run(device_module._resolve("127.0.0.1", 80)) == ("127.0.0.1", 80) + + +def test_send_failure_on_ping_and_setup_is_raised(net, monkeypatch): + """ping and setup fire one datagram and do not wait for a reply; a send + failure still has to reach the caller, as it did from a plain socket.""" + ep_holder = [] + original = net.__call__ + + async def go(): + + async def open_and_break(**kwargs): + transport, protocol = await original(**kwargs) + + def failing_sendto(data, addr=None): + protocol.error_received(OSError(101, "Network is unreachable")) + + transport.sendto = failing_sendto + ep_holder.append(transport) + return transport, protocol + + monkeypatch.setattr(device_module, "_open_endpoint", open_and_break) + with pytest.raises(OSError): + await broadlink.ping("192.0.2.1") + with pytest.raises(OSError): + await broadlink.setup("ssid", "pass", 3, ip_address="192.0.2.255") + return all(ep.closed for ep in ep_holder) + + assert run(go()) + + # ------------------------------------------------------------------ gendevice