refactor: extract firmware check logic into a dedicated helper method to reduce duplication (#527)

* refactor: extract firmware check logic into a dedicated helper method to reduce duplication

* handle JSON decode errors and non-dictionary responses from the GitHub API

Author: firstof9

openevsehttp/__main__.py

@@ -593,42 +593,11 @@ async def firmware_check(self) -> dict | None:
             return None
 
         try:
-            if self._session:
-                session = self._session
-                http_method = getattr(session, method)
-                _LOGGER.debug(
-                    "Connecting to %s using method %s",
-                    url,
-                    method,
-                )
-                async with http_method(url) as resp:
-                    if resp.status != 200:
-                        return None
-                    message = await resp.text()
-                    message = json.loads(message)
-                    response = {}
-                    response["latest_version"] = message["tag_name"]
-                    response["release_notes"] = message["body"]
-                    response["release_url"] = message["html_url"]
-                    return response
-            else:
+            if (session := self._session) is None:
                 async with aiohttp.ClientSession() as session:
-                    http_method = getattr(session, method)
-                    _LOGGER.debug(
-                        "Connecting to %s using method %s",
-                        url,
-                        method,
-                    )
-                    async with http_method(url) as resp:
-                        if resp.status != 200:
-                            return None
-                        message = await resp.text()
-                        message = json.loads(message)
-                        response = {}
-                        response["latest_version"] = message["tag_name"]
-                        response["release_notes"] = message["body"]
-                        response["release_url"] = message["html_url"]
-                        return response
+                    return await self._firmware_check_with_session(session, url, method)
+            else:
+                return await self._firmware_check_with_session(session, url, method)
 
         except (TimeoutError, ServerTimeoutError):
             _LOGGER.error("%s: %s", ERROR_TIMEOUT, url)
@@ -639,6 +608,33 @@ async def firmware_check(self) -> dict | None:
 
         return None
 
+    async def _firmware_check_with_session(
+        self, session: aiohttp.ClientSession, url: str, method: str
+    ) -> dict | None:
+        """Process a firmware check request with a given session."""
+        http_method = getattr(session, method)
+        _LOGGER.debug(
+            "Connecting to %s using method %s",
+            url,
+            method,
+        )
+        async with http_method(url) as resp:
+            if resp.status != 200:
+                return None
+            message = await resp.text()
+            try:
+                message = json.loads(message)
+            except json.JSONDecodeError:
+                _LOGGER.error("Failed to parse JSON response: %s", message)
+                return None
+
+            response = {}
+            if isinstance(message, dict):
+                response["latest_version"] = message.get("tag_name")
+                response["release_notes"] = message.get("body")
+                response["release_url"] = message.get("html_url")
+            return response
+
     def _version_check(self, min_version: str, max_version: str = "") -> bool:
         """Return bool if minimum version is met."""
         if "version" not in self._config:

tests/test_main.py

@@ -3174,6 +3174,14 @@ async def test_firmware_check_errors(mock_aioclient):
     )
     assert await charger.firmware_check() is None
 
+    # JSONDecodeError from github
+    mock_aioclient.get(url, status=200, body="not json")
+    assert await charger.firmware_check() is None
+
+    # Non-dict JSON from github
+    mock_aioclient.get(url, status=200, body='"just a string"')
+    assert await charger.firmware_check() == {}
+
 
 async def test_websocket_pong():
     """Test websocket handles pong message."""