refactor(connection): remove _resolve_discovery_fn test shim and document MH201 retry semantics

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 (

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:

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"