diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index e2c6081..5e33371 100644 --- a/sagemcom_api/client.py +++ b/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.""" diff --git a/tests/unit/test_request_retries.py b/tests/unit/test_request_retries.py new file mode 100644 index 0000000..a3eb65c --- /dev/null +++ b/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"]