Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(auth): should return false on bad authentication #9

Merged
merged 1 commit into from
Jan 6, 2024
Merged

fix(auth): should return false on bad authentication #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 76 additions & 25 deletions sunweg/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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
Expand Down Expand Up @@ -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"]

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(",", "."))

Expand All @@ -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.

Expand All @@ -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.

Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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
40 changes: 20 additions & 20 deletions tests/test_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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)
Expand All @@ -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(
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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"],
Expand All @@ -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