diff --git a/OWNd/connection/command_session.py b/OWNd/connection/command_session.py index 1245f2b..8a30b22 100644 --- a/OWNd/connection/command_session.py +++ b/OWNd/connection/command_session.py @@ -209,6 +209,10 @@ async def _locked_send( # gateway may have executed it even if its acknowledgement was lost. # Status requests are idempotent and can be retried safely after a # transport reset. + is_idempotent_query = is_status_request and ( + getattr(msg_obj, "_message_type", None) + in ("STATUS_REQUEST", "DIMENSION_REQUEST") + ) max_attempts = 2 for attempt in range(1, max_attempts + 1): @@ -304,7 +308,7 @@ async def _locked_send( # state yet: a retry that succeeds is an internal recycle and # must not flap the consumer (offline, then online again). await self._close_streams() - if attempt < max_attempts and (is_status_request or not written): + if attempt < max_attempts and (is_idempotent_query or not written): self._logger.debug( "%s Command session connection lost (%s), retrying once...", self._log_id, diff --git a/tests/test_connection.py b/tests/test_connection.py index 65e6085..8b06cda 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -336,6 +336,55 @@ async def reconnect() -> dict: assert retry_writer.written == [b"*#1*0##"] +@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_cancelled_command_closes_stream() -> None: session, writer = make_session(OWNCommandSession)