feat: make api testable

Author: mkuthan

custom_components/solis_cloud_control/__init__.py

@@ -7,7 +7,7 @@
 from custom_components.solis_cloud_control.coordinator import SolisCloudControlCoordinator
 
 from .api import SolisCloudControlApiClient
-from .const import CONF_INVERTER_SN
+from .const import API_BASE_URL, CONF_INVERTER_SN
 
 PLATFORMS: list[Platform] = [Platform.SELECT, Platform.TEXT, Platform.NUMBER, Platform.SWITCH, Platform.SENSOR]
 
@@ -18,7 +18,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     inverter_sn = entry.data[CONF_INVERTER_SN]
 
     session = aiohttp_client.async_get_clientsession(hass)
-    api_client = SolisCloudControlApiClient(api_key, api_token, session)
+    api_client = SolisCloudControlApiClient(API_BASE_URL, api_key, api_token, session)
 
     coordinator = SolisCloudControlCoordinator(hass, entry, api_client)
     entry.runtime_data = coordinator

custom_components/solis_cloud_control/api.py

@@ -4,12 +4,10 @@
 from datetime import datetime
 
 import aiohttp
-import backoff
 
 from custom_components.solis_cloud_control.utils import current_date, digest, format_date, sign_authorization
 
 from .const import (
-    API_BASE_URL,
     API_CONCURRENT_REQUESTS,
     API_CONTROL_ENDPOINT,
     API_READ_BATCH_ENDPOINT,
@@ -43,10 +41,12 @@ def __init__(
 class SolisCloudControlApiClient:
     def __init__(
         self,
+        base_url: str,
         api_key: str,
         api_token: str,
         session: aiohttp.ClientSession,
     ) -> None:
+        self._base_url = base_url
         self._api_key = api_key
         self._api_secret = api_token
         self._session = session
@@ -72,7 +72,7 @@ async def _request(self, date: datetime, endpoint: str, payload: dict[str, any]
             "Authorization": authorization,
         }
 
-        url = f"{API_BASE_URL}{endpoint}"
+        url = f"{self._base_url}{endpoint}"
 
         _LOGGER.debug("API request '%s': %s", endpoint, json.dumps(payload, indent=2))
 
@@ -99,18 +99,14 @@ async def _request(self, date: datetime, endpoint: str, payload: dict[str, any]
         except aiohttp.ClientError as err:
             raise SolisCloudControlApiError(f"Error accessing {url}: {str(err)}") from err
 
-    @backoff.on_exception(
-        backoff.constant,
-        SolisCloudControlApiError,
-        max_tries=API_RETRY_COUNT,
-        interval=API_RETRY_DELAY_SECONDS,
-        logger=_LOGGER,
-    )
     async def read(self, inverter_sn: str, cid: int) -> str:
         date = current_date()
         payload = {"inverterSn": inverter_sn, "cid": cid}
 
-        data = await self._request(date, API_READ_ENDPOINT, payload)
+        async def request() -> str:
+            return await self._request(date, API_READ_ENDPOINT, payload)
+
+        data = await _retry_request(request)
 
         if data is None:
             raise SolisCloudControlApiError("Read failed: 'data' field is missing in response")
@@ -120,18 +116,14 @@ async def read(self, inverter_sn: str, cid: int) -> str:
 
         return data["msg"]
 
-    @backoff.on_exception(
-        backoff.constant,
-        SolisCloudControlApiError,
-        max_tries=API_RETRY_COUNT,
-        interval=API_RETRY_DELAY_SECONDS,
-        logger=_LOGGER,
-    )
     async def read_batch(self, inverter_sn: str, cids: list[int]) -> dict[int, str]:
         date = current_date()
         payload = {"inverterSn": inverter_sn, "cids": ",".join(map(str, cids))}
 
-        data = await self._request(date, API_READ_BATCH_ENDPOINT, payload)
+        async def request() -> dict[str, any]:
+            return await self._request(date, API_READ_BATCH_ENDPOINT, payload)
+
+        data = await _retry_request(request)
 
         if data is None:
             raise SolisCloudControlApiError("ReadBatch failed: 'data' field is missing in response")
@@ -157,20 +149,16 @@ async def read_batch(self, inverter_sn: str, cids: list[int]) -> dict[int, str]:
 
         return result
 
-    @backoff.on_exception(
-        backoff.constant,
-        SolisCloudControlApiError,
-        max_tries=API_RETRY_COUNT,
-        interval=API_RETRY_DELAY_SECONDS,
-        logger=_LOGGER,
-    )
     async def control(self, inverter_sn: str, cid: int, value: str, old_value: str | None = None) -> None:
         date = current_date()
         payload = {"inverterSn": inverter_sn, "cid": cid, "value": value}
         if old_value is not None:
             payload["yuanzhi"] = old_value
 
-        data_array = await self._request(date, API_CONTROL_ENDPOINT, payload)
+        async def request() -> None:
+            return await self._request(date, API_CONTROL_ENDPOINT, payload)
+
+        data_array = _retry_request(request)
 
         if data_array is None:
             return
@@ -184,3 +172,17 @@ async def control(self, inverter_sn: str, cid: int, value: str, old_value: str |
                     raise SolisCloudControlApiError(f"Control failed: {error_msg}", response_code=str(code))
 
         return
+
+
+async def _retry_request(request: callable) -> any:
+    attempt = 0
+    while attempt < API_RETRY_COUNT:
+        try:
+            return await request()
+        except SolisCloudControlApiError as err:
+            attempt += 1
+            if attempt < API_RETRY_COUNT:
+                _LOGGER.warning("Retrying due to error: %s (attempt %d/%d)", str(err), attempt, API_RETRY_COUNT)
+                await asyncio.sleep(API_RETRY_DELAY_SECONDS)
+            else:
+                raise

custom_components/solis_cloud_control/config_flow.py

@@ -7,7 +7,7 @@
 
 from custom_components.solis_cloud_control.api import SolisCloudControlApiClient, SolisCloudControlApiError
 
-from .const import CID_STORAGE_MODE, CONF_INVERTER_SN, DOMAIN
+from .const import API_BASE_URL, CID_STORAGE_MODE, CONF_INVERTER_SN, DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -59,5 +59,5 @@ async def async_step_user(self, user_input: dict[str, any] | None = None) -> Con
 
     async def _test_credentials(self, api_key: str, api_token: str, inverter_sn: str) -> None:
         session = aiohttp_client.async_get_clientsession(self.hass)
-        api_client = SolisCloudControlApiClient(api_key, api_token, session)
+        api_client = SolisCloudControlApiClient(API_BASE_URL, api_key, api_token, session)
         await api_client.read(inverter_sn, CID_STORAGE_MODE)

custom_components/solis_cloud_control/manifest.json

@@ -1,14 +1,11 @@
 {
     "domain": "solis_cloud_control",
     "name": "Solis Cloud Control",
-    "version": "0.0.1",
+    "version": "0.0.2",
     "documentation": "https://github.com/mkuthan/solis-cloud-control",
     "codeowners": [
         "@mkuthan"
     ],
-    "requirements": [
-        "backoff==2.2.1"
-    ],
     "config_flow": true,
     "iot_class": "cloud_polling"
 }

pyproject.toml

@@ -1,16 +1,16 @@
 [project]
 name = "solis-cloud-control"
-version = "0.0.1"
+version = "0.0.2"
 requires-python = "~=3.13"
 dependencies = [
-    "homeassistant==2025.3.3", 
-    "backoff==2.2.1"
+    "homeassistant==2025.3.3",
 ]
 
 
 [tool.uv]
 dev-dependencies = [
     "pytest==8.3.3",
+    "pytest-aiohttp==1.1.0",
     "pytest-asyncio==0.25.3",
     "pytest-cov==6.0.0",
     "ruff==0.7.4",
@@ -20,7 +20,8 @@ dev-dependencies = [
 [tool.pytest.ini_options]
 pythonpath = "."
 testpaths = ["tests"]
-addopts = "--cov=example --cov-branch"
+addopts = "--cov=custom_components --cov-branch"
+asyncio_mode = "auto"
 
 [tool.ruff]
 line-length = 120
@@ -42,7 +43,11 @@ select = [
     "N",   # pep8-naming
 ]
 per-file-ignores = { "custom_components/**/*.py" = [
-    "ANN101",
-], "tests/*.py" = [
-    "ANN101",
-] }
+        "ANN101",
+    ], "tests/*.py" = [
+        "ANN001",
+        "ANN101",
+        "ANN201",
+        "ANN202",
+        "ARG001",
+    ]}

tests/test_api.py

@@ -0,0 +1,98 @@
+from unittest.mock import patch
+
+import pytest
+from aiohttp import web
+
+from custom_components.solis_cloud_control.api import SolisCloudControlApiClient, SolisCloudControlApiError
+from custom_components.solis_cloud_control.const import API_READ_ENDPOINT
+
+
+@pytest.fixture
+def create_api_client():
+    def _create_api_client(session):
+        return SolisCloudControlApiClient(
+            base_url="",
+            api_key="any key",
+            api_token="any token",
+            session=session,
+        )
+
+    return _create_api_client
+
+
+async def test_api_read(create_api_client, aiohttp_client):
+    any_inverter_sn = "any inverter sn"
+    any_cid = -1
+    any_result = "any result"
+
+    async def mock_read_endpoint(request):
+        body = await request.json()
+        assert body.get("inverterSn") == any_inverter_sn
+        assert body.get("cid") == any_cid
+        return web.json_response({"code": "0", "msg": "Success", "data": {"msg": any_result}})
+
+    app = web.Application()
+    app.router.add_route("POST", API_READ_ENDPOINT, mock_read_endpoint)
+
+    client = await aiohttp_client(app)
+    api_client = create_api_client(client)
+
+    result = await api_client.read(inverter_sn=any_inverter_sn, cid=any_cid)
+
+    assert result == any_result
+
+
+async def mock_read_endpoint_api_error(request):
+    return web.json_response({"code": "100", "msg": "API Error"})
+
+
+async def mock_read_endpoint_missing_data_field(request):
+    return web.json_response({"code": "0", "msg": "Success"})
+
+
+async def mock_read_endpoint_missing_msg_field(request):
+    return web.json_response({"code": "0", "msg": "Success", "data": {"unknown field": "any value"}})
+
+
+async def mock_read_endpoint_http_error(request):
+    return web.Response(status=500, text="Internal Server Error")
+
+
+@pytest.fixture
+def mock_retry_request():
+    async def no_retry_request(request):
+        return await request()
+
+    with patch("custom_components.solis_cloud_control.api._retry_request", new=no_retry_request):
+        yield
+
+
+@pytest.fixture(
+    params=[
+        (mock_read_endpoint_http_error, SolisCloudControlApiError("Internal Server Error", status_code=500)),
+        (
+            mock_read_endpoint_api_error,
+            SolisCloudControlApiError("API operation failed: API Error", response_code="100"),
+        ),
+        (
+            mock_read_endpoint_missing_data_field,
+            SolisCloudControlApiError("Read failed: 'data' field is missing in response"),
+        ),
+        (
+            mock_read_endpoint_missing_msg_field,
+            SolisCloudControlApiError("Read failed: 'msg' field is missing in response"),
+        ),
+    ]
+)
+async def test_api_read_errors(request, create_api_client, aiohttp_client, mock_retry_request) -> None:
+    mock_endpoint, expected_error = request.param
+    app = web.Application()
+    app.router.add_route("POST", API_READ_ENDPOINT, mock_endpoint)
+
+    client = await aiohttp_client(app)
+    api_client = create_api_client(client)
+
+    with pytest.raises(SolisCloudControlApiError) as excinfo:
+        await api_client.read(inverter_sn="any inverter", cid=-1)
+
+    assert str(excinfo.value) == str(expected_error)