Do not resend reboot or speed test after a dropped connection

sagemcom_api/client.py

@@ -213,8 +213,24 @@ def __get_response_value(self, response, index=0):
         (ClientConnectorError, ClientOSError, ServerDisconnectedError),
         max_tries=5,
     )
-    # pylint: disable=too-many-branches
     async def __post(self, url, data):
+        """Post a request that is safe to repeat, retrying on connection errors."""
+        return await self.__send(url, data)
+
+    @backoff.on_exception(
+        backoff.expo,
+        ClientConnectorError,
+        max_tries=5,
+    )
+    async def __post_once(self, url, data):
+        """Post a request that must not run twice, retrying only when the connection could not be opened.
+
+        A dropped connection after sending may mean the router already executed the action (e.g. a reboot).
+        """
+        return await self.__send(url, data)
+
+    # pylint: disable=too-many-branches
+    async def __send(self, url, data):
         async with self.session.post(url, data=data) as response:
             if response.status == 400:
                 result = await response.text()
@@ -279,8 +295,11 @@ async def __post(self, url, data):
 
             return result
 
-    async def __api_request_async(self, actions, priority=False):
-        """Build request to the internal JSON-req API."""
+    async def __api_request_async(self, actions, priority=False, idempotent=True):
+        """Build request to the internal JSON-req API.
+
+        :param idempotent: set to False for actions that must not be resent after a dropped connection
+        """
         self.__generate_request_id()
         self.__generate_nonce()
         self.__generate_auth_key()
@@ -300,7 +319,8 @@ async def __api_request_async(self, actions, priority=False):
 
         form_data = {"req": json.dumps(payload, separators=(",", ":"))}
         try:
-            result = await self.__post(api_host, form_data)
+            post = self.__post if idempotent else self.__post_once
+            result = await post(api_host, form_data)
             return result
         except (
             ClientConnectorError,
@@ -575,7 +595,7 @@ async def reboot(self):
             "parameters": {"source": "GUI"},
         }
 
-        response = await self.__api_request_async([action], False)
+        response = await self.__api_request_async([action], False, idempotent=False)
         data = self.__get_response_value(response)
 
         return data
@@ -590,7 +610,7 @@ async def run_speed_test(self, block_traffic: bool = False):
                 "parameters": {"BlockTraffic": block_traffic},
             }
         ]
-        return await self.__api_request_async(actions, False)
+        return await self.__api_request_async(actions, False, idempotent=False)
 
     async def get_speed_test_results(self) -> list[SpeedTestResult]:
         """Retrieve Speed Test results from Sagemcom F@st device."""

tests/unit/test_request_retries.py

@@ -0,0 +1,97 @@
+"""Tests for transport retries on actions that must not run twice."""
+
+import asyncio
+import json
+import urllib.parse
+from unittest.mock import AsyncMock
+
+import pytest
+from aiohttp import ClientSession, ClientTimeout
+
+from sagemcom_api.client import SagemcomClient
+from sagemcom_api.enums import EncryptionMethod
+
+OK_REPLY = {
+    "reply": {
+        "uid": 0,
+        "id": 0,
+        "error": {"code": 16777216, "description": "XMO_REQUEST_NO_ERR"},
+        "actions": [
+            {
+                "uid": 1,
+                "id": 0,
+                "error": {"code": 16777238, "description": "XMO_NO_ERR"},
+                "callbacks": [
+                    {"uid": 1, "result": {"code": 16777238, "description": "XMO_NO_ERR"}, "xpath": "Device", "parameters": {"value": "ok"}}
+                ],
+            }
+        ],
+        "events": [],
+    }
+}
+
+
+class FlakyRouter:
+    """Raw TCP server that accepts each request, then drops the first `drops` connections without answering."""
+
+    def __init__(self, drops: int):
+        """Drop the first `drops` connections after reading the request."""
+        self.drops = drops
+        self.methods: list[str] = []
+        self.server: asyncio.Server | None = None
+
+    async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        head = await reader.readuntil(b"\r\n\r\n")
+        length = next(int(line.split(b":")[1]) for line in head.split(b"\r\n") if line.lower().startswith(b"content-length"))
+        body = await reader.readexactly(length)
+        request = json.loads(urllib.parse.parse_qs(body.decode())["req"][0])["request"]
+        self.methods.extend(action["method"] for action in request["actions"])
+
+        if len(self.methods) <= self.drops:
+            writer.transport.abort()  # The router acted on the request, then went away without replying.
+            return
+
+        payload = json.dumps(OK_REPLY).encode()
+        writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n%s" % (len(payload), payload))
+        await writer.drain()
+        writer.close()
+
+    async def __aenter__(self) -> str:
+        """Start the server and return its host:port."""
+        self.server = await asyncio.start_server(self._handle, "127.0.0.1", 0)
+        port = self.server.sockets[0].getsockname()[1]
+        return f"127.0.0.1:{port}"
+
+    async def __aexit__(self, *exc) -> None:
+        """Stop the server."""
+        assert self.server is not None
+        self.server.close()
+        await self.server.wait_closed()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("call", "method"), [("reboot", "reboot"), ("run_speed_test", "speedTestClient")])
+async def test_non_idempotent_action_is_not_resent_after_disconnect(call, method):
+    """A reboot (or speed test) that reached the router must not be sent again when the connection drops."""
+    router = FlakyRouter(drops=99)
+    async with router as host, ClientSession(timeout=ClientTimeout(total=10)) as session:
+        client = SagemcomClient(host, "admin", "admin", EncryptionMethod.MD5, session=session)
+
+        with pytest.raises(ConnectionError):
+            await getattr(client, call)()
+
+    assert router.methods == [method]
+
+
+@pytest.mark.asyncio
+async def test_read_is_still_retried_after_disconnect(monkeypatch):
+    """Reads are safe to repeat, so a dropped connection is still retried."""
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())  # skip the backoff delay
+    router = FlakyRouter(drops=1)
+    async with router as host, ClientSession(timeout=ClientTimeout(total=10)) as session:
+        client = SagemcomClient(host, "admin", "admin", EncryptionMethod.MD5, session=session)
+
+        value = await client.get_value_by_xpath("Device")
+
+    assert value == "ok"
+    assert router.methods == ["getValue", "getValue"]