fix(auth): should return false on bad authentication

sunweg/api.py

@@ -46,7 +46,9 @@ def convert_situation_status(situation: int) -> Status:
     return Status.WARN
 
 
-def separate_value_metric(value_with_metric: str | None, default_metric: str = "") -> tuple[float, str]:
+def separate_value_metric(
+    value_with_metric: str | None, default_metric: str = ""
+) -> tuple[float, str]:
     """
     Separate the value from the metric.
 
@@ -62,12 +64,11 @@ def separate_value_metric(value_with_metric: str | None, default_metric: str = "
     split = value_with_metric.split(" ")
     return (
         float(split[0].replace(",", ".")),
-        default_metric if len(split) < 2 else split[1]
+        default_metric if len(split) < 2 else split[1],
     )
 
 
-
-class APIHelper():
+class APIHelper:
     """Class to call sunweg.net api."""
 
     SERVER_URI = SUNWEG_URL
@@ -120,7 +121,9 @@ def authenticate(self) -> bool:
             default=lambda o: o.__dict__,
         )
 
-        result = self._post(SUNWEG_LOGIN_PATH, user_data)
+        result = self._post(SUNWEG_LOGIN_PATH, user_data, False)
+        if not result["success"]:
+            return False
         self._token = result["token"]
         return result["success"]
 
@@ -168,7 +171,9 @@ def plant(self, id: int, retry=True) -> Plant | None:
         try:
             result = self._get(SUNWEG_PLANT_DETAIL_PATH + str(id))
 
-            (today_energy, today_energy_metric) = separate_value_metric(result["energiaGeradaHoje"], "kWh")
+            (today_energy, today_energy_metric) = separate_value_metric(
+                result["energiaGeradaHoje"], "kWh"
+            )
             total_power = separate_value_metric(result["AcumuladoPotencia"])[0]
             plant = Plant(
                 id=id,
@@ -220,8 +225,12 @@ def inverter(self, id: int, retry=True) -> Inverter | None:
         """
         try:
             result = self._get(SUNWEG_INVERTER_DETAIL_PATH + str(id))
-            (total_energy, total_energy_metric) = separate_value_metric(result["energiaacumulada"], "kWh")
-            (today_energy, today_energy_metric) = separate_value_metric(result["energiadodia"], "kWh")
+            (total_energy, total_energy_metric) = separate_value_metric(
+                result["energiaacumulada"], "kWh"
+            )
+            (today_energy, today_energy_metric) = separate_value_metric(
+                result["energiadodia"], "kWh"
+            )
             (power, power_metric) = separate_value_metric(result["potenciaativa"], "kW")
             inverter = Inverter(
                 id=id,
@@ -259,9 +268,17 @@ def complete_inverter(self, inverter: Inverter, retry=True) -> None:
         """
         try:
             result = self._get(SUNWEG_INVERTER_DETAIL_PATH + str(inverter.id))
-            (inverter.total_energy, inverter.total_energy_metric) = separate_value_metric(result["energiaacumulada"], "kWh")
-            (inverter.today_energy, inverter.today_energy_metric) = separate_value_metric(result["energiadodia"], "kWh")
-            (inverter.power, inverter.power_metric) = separate_value_metric(result["potenciaativa"], "kW")
+            (
+                inverter.total_energy,
+                inverter.total_energy_metric,
+            ) = separate_value_metric(result["energiaacumulada"], "kWh")
+            (
+                inverter.today_energy,
+                inverter.today_energy_metric,
+            ) = separate_value_metric(result["energiadodia"], "kWh")
+            (inverter.power, inverter.power_metric) = separate_value_metric(
+                result["potenciaativa"], "kW"
+            )
             inverter.power_factor = float(result["fatorpotencia"].replace(",", "."))
             inverter.frequency = float(result["frequencia"].replace(",", "."))
 
@@ -271,7 +288,14 @@ def complete_inverter(self, inverter: Inverter, retry=True) -> None:
                 self.authenticate()
                 self.complete_inverter(inverter, False)
 
-    def month_stats_production(self, year: int, month: int, plant: Plant, inverter: Inverter | None = None, retry: bool = True) -> list[ProductionStats]:
+    def month_stats_production(
+        self,
+        year: int,
+        month: int,
+        plant: Plant,
+        inverter: Inverter | None = None,
+        retry: bool = True,
+    ) -> list[ProductionStats]:
         """
         Retrieve month energy production statistics.
 
@@ -288,9 +312,18 @@ def month_stats_production(self, year: int, month: int, plant: Plant, inverter:
         :return: list of daily energy production statistics
         :rtype: list[ProductionStats]
         """
-        return self.month_stats_production_by_id(year, month, plant.id, inverter.id if inverter is not None else None, retry)
+        return self.month_stats_production_by_id(
+            year, month, plant.id, inverter.id if inverter is not None else None, retry
+        )
 
-    def month_stats_production_by_id(self, year: int, month: int, plant_id: int, inverter_id: int | None = None, retry: bool = True) -> list[ProductionStats]:
+    def month_stats_production_by_id(
+        self,
+        year: int,
+        month: int,
+        plant_id: int,
+        inverter_id: int | None = None,
+        retry: bool = True,
+    ) -> list[ProductionStats]:
         """
         Retrieve month energy production statistics.
 
@@ -307,14 +340,26 @@ def month_stats_production_by_id(self, year: int, month: int, plant_id: int, inv
         :return: list of daily energy production statistics
         :rtype: list[ProductionStats]
         """
-        inverter_str:str = str(inverter_id) if inverter_id is not None else ""
+        inverter_str: str = str(inverter_id) if inverter_id is not None else ""
         try:
-            result = self._get(SUNWEG_MONTH_STATS_PATH + f"idusina={plant_id}&idinversor={inverter_str}&date={format(month,'02')}/{year}")
-            return [ProductionStats(datetime.strptime(item["tempoatual"],"%Y-%m-%d").date(), float(item["energiapordia"]), float(item["prognostico"])) for item in result["graficomes"]]
+            result = self._get(
+                SUNWEG_MONTH_STATS_PATH
+                + f"idusina={plant_id}&idinversor={inverter_str}&date={format(month,'02')}/{year}"
+            )
+            return [
+                ProductionStats(
+                    datetime.strptime(item["tempoatual"], "%Y-%m-%d").date(),
+                    float(item["energiapordia"]),
+                    float(item["prognostico"]),
+                )
+                for item in result["graficomes"]
+            ]
         except LoginError:
             if retry:
                 self.authenticate()
-                return self.month_stats_production_by_id(year, month, plant_id, inverter_id, False)
+                return self.month_stats_production_by_id(
+                    year, month, plant_id, inverter_id, False
+                )
             return []
 
     def _populate_MPPT(self, result: dict, inverter: Inverter) -> None:
@@ -326,7 +371,9 @@ def _populate_MPPT(self, result: dict, inverter: Inverter) -> None:
                 string = String(
                     str_string["nome"],
                     float(result["inversor"]["leitura"][str_string["variaveltensao"]]),
-                    float(result["inversor"]["leitura"][str_string["variavelcorrente"]]),
+                    float(
+                        result["inversor"]["leitura"][str_string["variavelcorrente"]]
+                    ),
                     convert_situation_status(int(str_string["situacao"])),
                 )
                 mppt.strings.append(string)
@@ -346,27 +393,31 @@ def _populate_MPPT(self, result: dict, inverter: Inverter) -> None:
                 )
             )
 
-    def _get(self, path: str) -> dict:
+    def _get(self, path: str, launch_exception_on_error: bool = True) -> dict:
         """Do a get request returning a treated response."""
         res = self.session.get(self.SERVER_URI + path, headers=self._headers())
-        result = self._treat_response(res)
+        result = self._treat_response(res, launch_exception_on_error)
         return result
 
-    def _post(self, path: str, data: Any | None) -> dict:
+    def _post(
+        self, path: str, data: Any | None, launch_exception_on_error: bool = True
+    ) -> dict:
         """Do a post request returning a treated response."""
         res = self.session.post(
             self.SERVER_URI + path, data=data, headers=self._headers()
         )
-        result = self._treat_response(res)
+        result = self._treat_response(res, launch_exception_on_error)
         return result
 
-    def _treat_response(self, response: Response) -> dict:
+    def _treat_response(
+        self, response: Response, launch_exception_on_error: bool = True
+    ) -> dict:
         """Treat the response from requests."""
         if response.status_code == 401:
             raise LoginError("Request failed: %s" % response)
         if response.status_code != 200:
             raise SunWegApiError("Request failed: %s" % response)
         result = response.json()
-        if not result["success"]:
+        if launch_exception_on_error and not result["success"]:
             raise SunWegApiError(result["message"])
         return result

tests/test_api.py

@@ -8,7 +8,12 @@
 
 from requests import Response
 
-from sunweg.api import APIHelper, convert_situation_status, SunWegApiError, separate_value_metric
+from sunweg.api import (
+    APIHelper,
+    convert_situation_status,
+    SunWegApiError,
+    separate_value_metric,
+)
 from sunweg.device import Inverter, String
 from sunweg.util import Status
 
@@ -34,7 +39,6 @@ def setUp(self) -> None:
                     response._content = "".join(f.readlines()).encode()
                 self.responses[filename] = response
 
-
     def test_convert_situation_status(self) -> None:
         """Test the conversion from situation to status."""
         status_ok: Status = convert_situation_status(1)
@@ -45,46 +49,42 @@ def test_convert_situation_status(self) -> None:
         assert status_err == Status.ERROR
         assert status_wrn == Status.WARN
 
-
     def test_separate_value_metric_comma(self) -> None:
         """Test the separation from value and metric of string with comma."""
-        (value,metric) = separate_value_metric("0,0")
+        (value, metric) = separate_value_metric("0,0")
         assert value == 0
         assert metric == ""
-        (value,metric) = separate_value_metric("1,0", "W")
+        (value, metric) = separate_value_metric("1,0", "W")
         assert value == 1.0
         assert metric == "W"
-        (value,metric) = separate_value_metric("0,2 kW", "W")
+        (value, metric) = separate_value_metric("0,2 kW", "W")
         assert value == 0.2
         assert metric == "kW"
 
-
     def test_separate_value_metric_dot(self) -> None:
         """Test the separation from value and metric of string with dot."""
-        (value,metric) = separate_value_metric("0.0")
+        (value, metric) = separate_value_metric("0.0")
         assert value == 0
         assert metric == ""
-        (value,metric) = separate_value_metric("1.0", "W")
+        (value, metric) = separate_value_metric("1.0", "W")
         assert value == 1.0
         assert metric == "W"
-        (value,metric) = separate_value_metric("0.2 kW", "W")
+        (value, metric) = separate_value_metric("0.2 kW", "W")
         assert value == 0.2
         assert metric == "kW"
 
-
     def test_separate_value_metric_none_int(self) -> None:
         """Test the separation from value and metric of string with dot."""
-        (value,metric) = separate_value_metric(None)
+        (value, metric) = separate_value_metric(None)
         assert value == 0
         assert metric == ""
-        (value,metric) = separate_value_metric("1", "W")
+        (value, metric) = separate_value_metric("1", "W")
         assert value == 1.0
         assert metric == "W"
-        (value,metric) = separate_value_metric("2 kW", "W")
+        (value, metric) = separate_value_metric("2 kW", "W")
         assert value == 2.0
         assert metric == "kW"
 
-
     def test_error500(self) -> None:
         """Test error 500."""
         with patch(
@@ -112,9 +112,7 @@ def test_authenticate_failed(self) -> None:
             return_value=self.responses["auth_fail_response.json"],
         ):
             api = APIHelper("[email protected]", "password")
-            with pytest.raises(SunWegApiError) as e_info:
-                api.authenticate()
-            assert e_info.value.__str__() == "Error message"
+            assert not api.authenticate()
 
     def test_list_plants_none_success(self) -> None:
         """Test list plants with empty plant list."""
@@ -369,7 +367,7 @@ def test_month_stats_401(self) -> None:
             assert len(stats) == 0
 
     def test_month_stats_success(self) -> None:
-        """Test month stats with data from server.""" 
+        """Test month stats with data from server."""
         with patch(
             "requests.Session.get",
             return_value=self.responses["month_stats_success_response.json"],
@@ -384,5 +382,7 @@ def test_month_stats_success(self) -> None:
                 assert stat.date == date(2023, 12, i)
                 assert isinstance(stat.production, float)
                 assert stat.prognostic == 98.774193548387
-                assert stat.__str__().startswith("<class 'sunweg.util.ProductionStats'>")
+                assert stat.__str__().startswith(
+                    "<class 'sunweg.util.ProductionStats'>"
+                )
                 i += 1