Improve NextLocation warnings for user

Author: CoMPaTech

custom_components/stromer/__init__.py

@@ -10,7 +10,7 @@
 
 from .const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, DOMAIN, LOGGER
 from .coordinator import StromerDataUpdateCoordinator
-from .stromer import ApiError, Stromer
+from .stromer import ApiError, NextLocationError, Stromer
 
 SCAN_INTERVAL = timedelta(minutes=10)
 
@@ -42,6 +42,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         await stromer.stromer_connect()
     except ApiError as ex:
         raise ConfigEntryNotReady("Error while communicating to Stromer API") from ex
+    except NextLocationError as ex:
+        raise ConfigEntryNotReady("Error while getting authentication location %s", ex) from ex
 
     # Ensure migration from v3 single bike
     if "bike_id" not in entry.data:

custom_components/stromer/config_flow.py

@@ -12,7 +12,7 @@
 from homeassistant.exceptions import HomeAssistantError
 
 from .const import BIKE_DETAILS, CONF_CLIENT_ID, CONF_CLIENT_SECRET, DOMAIN, LOGGER
-from .stromer import Stromer
+from .stromer import ApiError, NextLocationError, Stromer
 
 STEP_USER_DATA_SCHEMA = vol.Schema(
     {
@@ -32,9 +32,17 @@ async def validate_input(_: HomeAssistant, data: dict[str, Any]) -> dict:
     client_secret = data.get(CONF_CLIENT_SECRET, None)
 
     # Initialize connection to stromer to validate credentials
-    stromer = Stromer(username, password, client_id, client_secret)
-    if not await stromer.stromer_connect():
+    try:
+        stromer = Stromer(username, password, client_id, client_secret)
+        connected: bool = await stromer.stromer_connect()
+    except ApiError as ex:
+        raise CannotConnect("Error while connecting to Stromer API %s", ex) from ex
+    except NextLocationError as ex:
+        raise CannotConnect("Error while getting authentication location %s", ex) from ex
+
+    if not connected:
         raise InvalidAuth
+
     LOGGER.debug("Credentials validated successfully")
 
     # All bikes information available

custom_components/stromer/coordinator.py

@@ -6,7 +6,7 @@
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import DOMAIN, LOGGER
-from .stromer import ApiError, Stromer
+from .stromer import ApiError, NextLocationError, Stromer
 
 
 class StromerData(NamedTuple):
@@ -42,8 +42,10 @@ async def _async_update_data(self) -> StromerData:
             data = [bike_data, self.stromer.bike_id, self.stromer.bike_name]
             LOGGER.debug("Stromer data %s updated", data)
 
-        except ApiError as err:
-            raise UpdateFailed(f"Error communicating with API: {err}") from err
-        except Exception as err:
-            raise ConfigEntryAuthFailed from err
+        except ApiError as ex:
+            raise UpdateFailed(f"Error communicating with API: {ex}") from ex
+        except NextLocationError as ex:
+            raise UpdateFailed("Error while getting authentication location %s", ex) from ex
+        except Exception as ex:
+            raise ConfigEntryAuthFailed from ex
         return StromerData(*data)  # type: ignore [arg-type]

custom_components/stromer/manifest.json

@@ -8,5 +8,5 @@
   "iot_class": "cloud_polling",
   "issue_tracker": "https://github.com/CoMPaTech/stromer/issues",
   "requirements": [],
-  "version": "0.4.3a1"
+  "version": "0.4.3a2"
 }

custom_components/stromer/stromer.py

@@ -170,16 +170,13 @@ async def stromer_get_code(self) -> None:
             url, data=data, headers={"Referer": url}, allow_redirects=False
         )
         next_loc = res.headers.get("Location")
-        LOGGER.error("DEBUG location result %s", await res.text())
-        LOGGER.error("DEBUG determined next location: %s", next_loc)
 
         next_url = f"{self.base_url}{next_loc}"
-        LOGGER.error("DEBUG next url: %s", next_url)
+        if not (next_loc.startswith("/") or next_loc.startswith("?")):
+            raise NextLocationError(f"Invalid next location: '{next_loc}'. Expected start with '/' or '?'.")
 
         res = await self._websession.get(next_url, allow_redirects=False)
         self._code = res.headers.get("Location")
-        LOGGER.error("DEBUG code result %s", await res.text())
-        LOGGER.error("DEBUG code determined: %s", self._code)
         self._code = self._code.split("=")[1]  # type: ignore[union-attr]
 
     async def stromer_get_access_token(self) -> None:
@@ -265,3 +262,6 @@ async def stromer_call_api(self, endpoint: str, full=False) -> Any:
 
 class ApiError(Exception):
     """Error to indicate something wrong with the API."""
+
+class NextLocationError(Exception):
+    """Error to indicate something wrong returned in next location."""

pyproject.toml

@@ -69,7 +69,7 @@ lint.select = [
     "TID251", # Banned imports
     "TRY004", # Prefer TypeError exception for invalid type
     "B904", # Use raise from to specify exception cause  # warning: `TRY200` has been remapped to `B904`.
-    "TRY302", # Remove exception handler; error is immediately re-raised
+    "TRY203", # Remove exception handler; error is immediately re-raised
     "UP",  # pyupgrade
     "W",  # pycodestyle
 ]
@@ -93,7 +93,6 @@ lint.ignore = [
     "UP006", # keep type annotation style as is
     "UP007", # keep type annotation style as is
     # Ignored due to performance: https://github.com/charliermarsh/ruff/issues/2923
-    "UP038", # Use `X | Y` in `isinstance` call instead of `(X, Y)`
 ]
 
 [tool.ruff.lint.flake8-import-conventions.extend-aliases]