diff --git a/OWNd/connection/command_session.py b/OWNd/connection/command_session.py index dfcfb17..111bf47 100644 --- a/OWNd/connection/command_session.py +++ b/OWNd/connection/command_session.py @@ -271,6 +271,17 @@ async def _locked_send( # single fast connect attempt (and a short cooldown on failure) so # queued callers behind _send_lock never stall for a full 5-attempt # exponential backoff loop. + # + # Note on single-slot gateways (e.g. MH201, which accepts TCP and + # closes during negotiation when its session slot is exhausted; see + # OpenWebNet-HA/OWNd#132 and OpenWebNet-HA/MyHOME#758): callers that + # recycle idle command sockets (such as MyHOME's + # CommandWorkerPool.sending_loop() after COMMAND_SESSION_IDLE_TIMEOUT) + # invoke `await session.connect()` directly when `not session.is_open` + # before calling `send()`, preserving the full MAX_CONNECT_ATTEMPTS=5 + # (~13.3 s) retry window on session open. This single-attempt path + # only runs when `send()` is called directly on a closed session or + # on attempt 2 after a mid-flight transport drop. if self._stream_reader is None or self._stream_writer is None: now = time.monotonic() if ( diff --git a/OWNd/connection/gateway.py b/OWNd/connection/gateway.py index 14e0355..996c684 100644 --- a/OWNd/connection/gateway.py +++ b/OWNd/connection/gateway.py @@ -4,23 +4,12 @@ from collections.abc import Mapping import logging -import sys from typing import Any from urllib.parse import urlparse from ..discovery import find_gateways, get_gateway, get_port from ..profiles import GatewayProfile, get_gateway_profile -_ORIG_FIND_GATEWAYS = find_gateways -_ORIG_GET_GATEWAY = get_gateway -_ORIG_GET_PORT = get_port - - -def _resolve_discovery_fn(name: str, local_fn: Any, orig_fn: Any) -> Any: - """Resolve discovery helper, honoring overrides on either gateway or connection.""" - pkg = sys.modules.get(__package__ or "") - return getattr(pkg, name, local_fn) if local_fn is orig_fn else local_fn - def _first_scalar(value: Any, default: Any = None) -> Any: """Return a scalar from legacy tuple/list discovery values.""" @@ -128,8 +117,7 @@ def log_id(self, value: str) -> None: async def get_first_available_gateway( cls, password: str | None = None ) -> OWNGateway | None: - finder = _resolve_discovery_fn("find_gateways", find_gateways, _ORIG_FIND_GATEWAYS) - local_gateways = await finder() + local_gateways = await find_gateways() if not local_gateways: return None local_gateways[0]["password"] = password @@ -138,8 +126,7 @@ async def get_first_available_gateway( @classmethod async def find_from_address(cls, address: str | None) -> OWNGateway | None: if address is not None: - getter = _resolve_discovery_fn("get_gateway", get_gateway, _ORIG_GET_GATEWAY) - gateway = await getter(address) + gateway = await get_gateway(address) return cls(gateway) if gateway is not None else None return await cls.get_first_available_gateway() @@ -163,8 +150,7 @@ async def build_from_discovery_info( "ssdp_location" in discovery_info and discovery_info["ssdp_location"] is not None ): - port_getter = _resolve_discovery_fn("get_port", get_port, _ORIG_GET_PORT) - discovery_info["port"] = await port_getter(discovery_info["ssdp_location"]) + discovery_info["port"] = await get_port(discovery_info["ssdp_location"]) elif "address" in discovery_info and discovery_info["address"] is not None: return await cls.find_from_address(discovery_info["address"]) else: diff --git a/tests/test_connection.py b/tests/test_connection.py index 4962096..a8aa584 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -682,55 +682,6 @@ async def test_command_session_rejects_invalid_or_empty_message() -> None: assert logger.error.call_count == 6 -@pytest.mark.asyncio -@pytest.mark.parametrize( - "frame", - [ - "*1*1*11##", - "*#4*000*0*0215##", - "*#4*000*#0*0215##", - ], -) -async def test_written_non_query_with_status_request_flag_is_not_replayed_after_lost_ack( - frame: str, -) -> None: - session, writer = make_session(OWNCommandSession) - assert isinstance(session, OWNCommandSession) - assert session._stream_reader is not None - session._stream_reader.feed_eof() - - with patch.object(session, "connect", new_callable=AsyncMock) as connect: - result = await session.send(frame, is_status_request=True) - - assert result is None - assert writer.written == [frame.encode()] - connect.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_disconnected_dimension_request_retries_once() -> None: - session, original_writer = make_session(OWNCommandSession) - assert isinstance(session, OWNCommandSession) - assert session._stream_reader is not None - session._stream_reader.feed_eof() - - retry_writer = FakeWriter() - - async def reconnect() -> dict[str, object]: - session._stream_reader = asyncio.StreamReader() - session._stream_reader.feed_eof() - session._stream_writer = retry_writer # type: ignore[assignment] - return {"Success": True, "Message": None} - - with patch.object(session, "connect", side_effect=reconnect) as connect: - result = await session.send("*#4*000*0##", is_status_request=True) - - assert result is None - connect.assert_awaited_once() - assert original_writer.written == [b"*#4*000*0##"] - assert retry_writer.written == [b"*#4*000*0##"] - - @pytest.mark.asyncio async def test_typed_exceptions_and_last_error() -> None: from OWNd import ( @@ -887,6 +838,10 @@ def broken_cb(err: OWNAuthError) -> None: session.on_auth_error = broken_cb assert session.on_auth_error is broken_cb session._notify_auth_error(OWNAuthError("bad pw", reason="password_error")) + saved_logger = session._logger + session._logger = None # type: ignore[assignment] + session._notify_auth_error(OWNAuthError("bad pw", reason="password_error")) + session._logger = saved_logger # test_connection(raise_on_error=True) refused after 3 attempts with ( @@ -915,7 +870,9 @@ def broken_cb(err: OWNAuthError) -> None: ): with pytest.raises(OWNAuthError): await session.test_connection(raise_on_error=True) - assert len(auth_called) == 1 + res_no_raise = await session.test_connection(raise_on_error=False) + assert res_no_raise == {"Success": False, "Message": "password_error"} + assert len(auth_called) == 2 with ( patch( @@ -931,6 +888,21 @@ def broken_cb(err: OWNAuthError) -> None: with pytest.raises(OWNNackError): await session.test_connection(raise_on_error=True) + # connect(raise_on_error=True) with fatal non-auth negotiation error + with ( + patch( + "OWNd.connection.session.asyncio.open_connection", + new=AsyncMock(return_value=(MagicMock(), MagicMock())), + ), + patch.object( + session, + "_negotiate", + new=AsyncMock(return_value={"Success": False, "Message": "negotiation_error"}), + ), + ): + with pytest.raises(OWNProtocolError): + await session.connect(raise_on_error=True) + # test_connection(raise_on_error=True) ConnectionResetError and TimeoutError during _negotiate with ( patch( @@ -1009,6 +981,12 @@ def broken_cb(err: OWNAuthError) -> None: assert isinstance(ev_sess, OWNEventSession) ev_sess._stream_reader = None ev_sess._stream_writer = None + with patch.object( + ev_sess, + "_reconnect", + new=AsyncMock(return_value={"Success": False, "Message": "connection_error"}), + ): + assert await ev_sess.get_next(raise_on_auth_error=True) is None with patch.object( ev_sess, "_reconnect", @@ -1017,8 +995,4 @@ def broken_cb(err: OWNAuthError) -> None: with pytest.raises(OWNAuthError): await ev_sess.get_next(raise_on_auth_error=True) - # Legacy patch("OWNd.connection.find_gateways") compatibility on OWNGateway - with patch("OWNd.connection.find_gateways", new=AsyncMock(return_value=[{"address": "192.168.1.77"}])): - gw = await OWNGateway.get_first_available_gateway() - assert gw is not None and gw.host == "192.168.1.77"