diff --git a/README.md b/README.md index 9398143..e30c9f6 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,26 @@ pip install oligo[requests] pip install oligo[asyncio] ``` +### Autenticación + +Puedes pasar el usuario y la contraseña directamente o usar las variables de entorno `I_DE_USER` e `I_DE_PASSWORD`: + +```python +from oligo import Iber + +connection = Iber() +connection.login("user", "password") +``` + +O usando variables de entorno: + +```python +from oligo import Iber + +connection = Iber() +connection.login() # Lee I_DE_USER e I_DE_PASSWORD del entorno +``` + ### Ejemplos: #### Consultar consumo actual (Sync): @@ -118,6 +138,44 @@ async def main(): asyncio.run(main()) ``` +#### Obtener el consumo horario facturado durante un periodo (Sync) + +```python +from oligo import Iber +from datetime import date, timedelta + +connection = Iber() +connection.login("user", "password") + +from_date = date.today() - timedelta(days=7) +until_date = date.today() - timedelta(days=1) + +consumo = connection.billed_consumption(from_date, until_date) + +print(consumo[:10]) +``` + +#### Obtener el consumo horario facturado durante un periodo (ASync) + +```python +import asyncio +from oligo.asyncio import AsyncIber +from datetime import date, timedelta + +async def main(): + connection = AsyncIber() + await connection.login("user", "password") + + from_date = date.today() - timedelta(days=7) + until_date = date.today() - timedelta(days=1) + + consumo = await connection.billed_consumption(from_date, until_date) + + print(consumo[:10]) + +asyncio.run(main()) +``` + Los datos son el consumo por hora en Watt-horas. En este caso tendremos los dato de una semana, que son 7 por 24, 168 valores. Si sumamos y dividimos por 1000, tenemos el consumo de una semana en kWh. @@ -129,6 +187,26 @@ por 1000, tenemos el consumo de una semana en kWh. > No new features will be added, only bugs will be fixed while the i-DE web api > continues to work in the same way. +### Authentication + +You can pass the username and password directly or use the `I_DE_USER` and `I_DE_PASSWORD` environment variables: + +```python +from oligo import Iber + +connection = Iber() +connection.login("user", "password") +``` + +Or using environment variables: + +```python +from oligo import Iber + +connection = Iber() +connection.login() # Reads I_DE_USER and I_DE_PASSWORD from environment +``` + ### Install: ``` @@ -225,6 +303,44 @@ async def main(): asyncio.run(main()) ``` +#### Retrieve the hourly billed consumption during a time period (Sync) + +```python +from oligo import Iber +from datetime import date, timedelta + +connection = Iber() +connection.login("user", "password") + +from_date = date.today() - timedelta(days=7) +until_date = date.today() - timedelta(days=1) + +consumo = connection.billed_consumption(from_date, until_date) + +print(consumo[:10]) +``` + +#### Retrieve the hourly billed consumption during a time period (Async) + +```python +import asyncio +from oligo.asyncio import AsyncIber +from datetime import date, timedelta + +async def main(): + connection = AsyncIber() + await connection.login("user", "password") + + from_date = date.today() - timedelta(days=7) + until_date = date.today() - timedelta(days=1) + + consumo = await connection.billed_consumption(from_date, until_date) + + print(consumo[:10]) + +asyncio.run(main()) +``` + The values are the consumption in Watt-hours. In this case, we have the data of one week, which are 7 times 24, 168 values. If we sum and divide by 1000, we will have the total consumption from one week in kWh. diff --git a/oligo/asyncio/asynciber.py b/oligo/asyncio/asynciber.py index 9ffde55..6ae5e48 100644 --- a/oligo/asyncio/asynciber.py +++ b/oligo/asyncio/asynciber.py @@ -1,7 +1,14 @@ +import os + from deprecated.classic import deprecated -from ..exception import SessionException, ResponseException, NoResponseException, LoginException, \ - SelectContractException +from ..exception import ( + SessionException, + ResponseException, + NoResponseException, + LoginException, + SelectContractException, +) try: import aiohttp @@ -23,6 +30,7 @@ BORRAR_ESCENARIO_URL = "escenarioNew/borrarEscenario/" OBTENER_PERIODO_URL = "consumoNew/obtenerDatosConsumoPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" OBTENER_PERIODO_GENERACION_URL = "consumoNew/obtenerDatosGeneracionPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" +OBTENER_PERIODO_FACTURADO_URL = "consumoNew/obtenerDatosConsumoFacturado/numFactura/null//fechaDesde//{}00:00:00//fechaHasta//{}23:59:00/true/" class AsyncIber: @@ -59,8 +67,26 @@ async def __request( raise NoResponseException return data - async def login(self, user: str, password: str) -> bool: - """Creates session with your credentials""" + async def login( + self, user: Optional[str] = None, password: Optional[str] = None + ) -> bool: + """Creates session with your credentials. + + If ``user`` or ``password`` are not provided, their values are read + from the ``I_DE_USER`` and ``I_DE_PASSWORD`` environment variables. + Explicit arguments take precedence over environment variables. + """ + user = user or os.getenv("I_DE_USER") + password = password or os.getenv("I_DE_PASSWORD") + if not user or not password: + raise LoginException( + user or "unknown", + message=( + "Login failed: user and password are required. " + "Set I_DE_USER and I_DE_PASSWORD environment variables " + "or pass them as arguments." + ), + ) self.__session = aiohttp.ClientSession() payload = [ user, @@ -197,3 +223,35 @@ async def production(self, start: datetime, end: datetime) -> list: async def total_consumption(self, start, end) -> float: data = await self._consumption_raw(start, end) return float(data["acumulado"]) + + async def _consumption_facturado_raw(self, start: datetime, end: datetime) -> list: + return await self.__request( + OBTENER_PERIODO_FACTURADO_URL.format( + start.strftime("%d-%m-%Y"), end.strftime("%d-%m-%Y") + ) + ) + + # Get billed consumption data from a time period + # + # start/end: datetime.date objects indicating the time period (both inclusive) + # + # Returns a list of billed consumptions starting at midnight on the start day until 23:00 on the last day. + # Each value is the hourly billed consumption in Wh. + async def billed_consumption(self, start: datetime, end: datetime) -> list: + data = await self._consumption_facturado_raw(start, end) + return [float(x["valor"]) for x in data["y"]["data"][0] if x] + + # Get total billed consumption in Wh (Watt-hour) over a time period + # + # start/end: datetime.date objects indicating the time period (both inclusive) + async def total_billed_consumption(self, start, end) -> float: + data = await self._consumption_facturado_raw(start, end) + return float(data["acumulado"]) + + @deprecated("Use 'billed_consumption' method instead") + async def consumption_facturado(self, start: datetime, end: datetime) -> list: + return await self.billed_consumption(start, end) + + @deprecated("Use 'total_billed_consumption' method instead") + async def total_consumption_facturado(self, start, end) -> float: + return await self.total_billed_consumption(start, end) diff --git a/oligo/exception.py b/oligo/exception.py index f1ed756..8e9218d 100644 --- a/oligo/exception.py +++ b/oligo/exception.py @@ -8,13 +8,15 @@ def __init__(self, status_code): class LoginException(IberException): - def __init__(self, username): - super().__init__(f'Unable to log in with user {username}') + def __init__(self, username, message=None): + if message is None: + message = f"Unable to log in with user {username}" + super().__init__(message) class SessionException(IberException): def __init__(self): - super().__init__('Session required, use login() method to obtain a session') + super().__init__("Session required, use login() method to obtain a session") class NoResponseException(IberException): diff --git a/oligo/requests/iber.py b/oligo/requests/iber.py index e6db7d3..31d86a4 100644 --- a/oligo/requests/iber.py +++ b/oligo/requests/iber.py @@ -1,3 +1,4 @@ +import os from datetime import datetime from deprecated.classic import deprecated @@ -7,25 +8,49 @@ except ImportError: raise RuntimeError("Iber requires requests module") -from ..exception import LoginException, ResponseException, NoResponseException, SelectContractException, \ - SessionException +from ..exception import ( + LoginException, + ResponseException, + NoResponseException, + SelectContractException, + SessionException, +) class Iber: - __domain = "https://www.i-de.es" __login_url = __domain + "/consumidores/rest/loginNew/login" - __watthourmeter_url = __domain + "/consumidores/rest/escenarioNew/obtenerMedicionOnline/24" + __watthourmeter_url = ( + __domain + "/consumidores/rest/escenarioNew/obtenerMedicionOnline/24" + ) __icp_status_url = __domain + "/consumidores/rest/rearmeICP/consultarEstado" __contracts_url = __domain + "/consumidores/rest/cto/listaCtos/" __contract_detail_url = __domain + "/consumidores/rest/detalleCto/detalle/" __contract_selection_url = __domain + "/consumidores/rest/cto/seleccion/" - __obtener_escenarios_url = __domain + "/consumidores/rest/escenarioNew/obtenerEscenariosRest/" - __obtener_escenario_url = __domain + "/consumidores/rest/escenarioNew/refrescarEscenario/" - __guardar_escenario_url = __domain + "/consumidores/rest/escenarioNew/confirmarMedicionOnLine/{}/1/{}" - __borrar_escenario_url = __domain + "/consumidores/rest/escenarioNew/borrarEscenario/" - __obtener_periodo_url = __domain + "/consumidores/rest/consumoNew/obtenerDatosConsumoPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" # date format: 07-11-2020 - that's 7 Nov 2020 - __obtener_periodo_generacion_url = __domain + "/consumidores/rest/consumoNew/obtenerDatosGeneracionPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" # date format: 07-11-2020 - that's 7 Nov 2020 + __obtener_escenarios_url = ( + __domain + "/consumidores/rest/escenarioNew/obtenerEscenariosRest/" + ) + __obtener_escenario_url = ( + __domain + "/consumidores/rest/escenarioNew/refrescarEscenario/" + ) + __guardar_escenario_url = ( + __domain + "/consumidores/rest/escenarioNew/confirmarMedicionOnLine/{}/1/{}" + ) + __borrar_escenario_url = ( + __domain + "/consumidores/rest/escenarioNew/borrarEscenario/" + ) + __obtener_periodo_url = ( + __domain + + "/consumidores/rest/consumoNew/obtenerDatosConsumoPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" + ) # date format: 07-11-2020 - that's 7 Nov 2020 + __obtener_periodo_generacion_url = ( + __domain + + "/consumidores/rest/consumoNew/obtenerDatosGeneracionPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" + ) # date format: 07-11-2020 - that's 7 Nov 2020 + __obtener_periodo_facturado_url = ( + __domain + + "/consumidores/rest/consumoNew/obtenerDatosConsumoFacturado/numFactura/null//fechaDesde//{}00:00:00//fechaHasta//{}23:59:00/true/" + ) __headers = { "Content-Type": "application/json; charset=utf-8", @@ -43,9 +68,27 @@ def __init__(self, session=None): """Iber class __init__ method.""" self.__session = session - def login(self, user, password, session=Session()): - """Creates session with your credentials""" + def login(self, user=None, password=None, session=None): + """Creates session with your credentials. + + If ``user`` or ``password`` are not provided, their values are read + from the ``I_DE_USER`` and ``I_DE_PASSWORD`` environment variables. + Explicit arguments take precedence over environment variables. + """ + if session is None: + session = Session() self.__session = session + user = user or os.getenv("I_DE_USER") + password = password or os.getenv("I_DE_PASSWORD") + if not user or not password: + raise LoginException( + user or "unknown", + message=( + "Login failed: user and password are required. " + "Set I_DE_USER and I_DE_PASSWORD environment variables " + "or pass them as arguments." + ), + ) login_data = [ user, password, @@ -58,7 +101,9 @@ def login(self, user, password, session=Session()): "s", "", ] - response = self.__session.request("POST", self.__login_url, headers=self.__headers, json=login_data) + response = self.__session.request( + "POST", self.__login_url, headers=self.__headers, json=login_data + ) if response.status_code != 200: self.__session = None raise ResponseException(response.status_code) @@ -74,18 +119,20 @@ def __check_session(self): def measurement(self): """Returns a measurement from the powermeter.""" self.__check_session() - response = self.__session.request("GET", self.__watthourmeter_url, headers=self.__headers) + response = self.__session.request( + "GET", self.__watthourmeter_url, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() return { - "id": json_response['codSolicitudTGT'], + "id": json_response["codSolicitudTGT"], "meter": json_response["valLecturaContador"], - "consumption": json_response['valMagnitud'], - "icp": json_response['valInterruptor'], - "raw_response": json_response + "consumption": json_response["valMagnitud"], + "icp": json_response["valInterruptor"], + "raw_response": json_response, } def current_kilowatt_hour_read(self): @@ -94,7 +141,7 @@ def current_kilowatt_hour_read(self): def current_power_consumption(self): """Returns your current power consumption.""" - return self.measurement()['consumption'] + return self.measurement()["consumption"] @deprecated("Use 'current_power_consumption' method instead") def watthourmeter(self): @@ -104,7 +151,9 @@ def watthourmeter(self): def icpstatus(self): """Returns the status of your ICP.""" self.__check_session() - response = self.__session.request("POST", self.__icp_status_url, headers=self.__headers) + response = self.__session.request( + "POST", self.__icp_status_url, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -117,7 +166,9 @@ def icpstatus(self): def contracts(self): self.__check_session() - response = self.__session.request("GET", self.__contracts_url, headers=self.__headers) + response = self.__session.request( + "GET", self.__contracts_url, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -128,7 +179,9 @@ def contracts(self): def contract(self): self.__check_session() - response = self.__session.request("GET", self.__contract_detail_url, headers=self.__headers) + response = self.__session.request( + "GET", self.__contract_detail_url, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -137,7 +190,9 @@ def contract(self): def contractselect(self, id): self.__check_session() - response = self.__session.request("GET", self.__contract_selection_url + id, headers=self.__headers) + response = self.__session.request( + "GET", self.__contract_selection_url + id, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -148,63 +203,79 @@ def contractselect(self, id): def scene_list(self): self.__check_session() - response = self.__session.request("GET", self.__obtener_escenarios_url, headers=self.__headers) + response = self.__session.request( + "GET", self.__obtener_escenarios_url, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() return { - "scene_names": json_response['y']['smps'], - "raw_response" : json_response + "scene_names": json_response["y"]["smps"], + "raw_response": json_response, } def scene_get(self, name): self.__check_session() - get_data = "{{\"nomEscenario\":\"{}\"}}".format(name) - response = self.__session.request("POST", self.__obtener_escenario_url, data=get_data, headers=self.__headers) + get_data = '{{"nomEscenario":"{}"}}'.format(name) + response = self.__session.request( + "POST", self.__obtener_escenario_url, data=get_data, headers=self.__headers + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() return { - "name": json_response['nomEscenario'], - "description": json_response['descripcion'], - "consumption": json_response['numLcaInsta'], - "raw_response" : json_response + "name": json_response["nomEscenario"], + "description": json_response["descripcion"], + "consumption": json_response["numLcaInsta"], + "raw_response": json_response, } - def scene_save(self, consumption, measurement_id, description): self.__check_session() name = datetime.now().strftime("%d/%m/%Y %H:%M:%S") - save_data = "{{\"nomEscenario\":\"{}\",\"descripcion\":\"{}\"}}".format(name, description) - response = self.__session.request("POST", self.__guardar_escenario_url.format(consumption, measurement_id), data=save_data, headers=self.__headers) + save_data = '{{"nomEscenario":"{}","descripcion":"{}"}}'.format( + name, description + ) + response = self.__session.request( + "POST", + self.__guardar_escenario_url.format(consumption, measurement_id), + data=save_data, + headers=self.__headers, + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() - return { - "name": json_response['nomEscenario'], - "raw_response" : json_response - } + return {"name": json_response["nomEscenario"], "raw_response": json_response} def scene_delete(self, name): self.__check_session() - delete_data = "{{\"nomEscenario\":\"{}\"}}".format(name) - response = self.__session.request("POST", self.__borrar_escenario_url, data=delete_data, headers=self.__headers) + delete_data = '{{"nomEscenario":"{}"}}'.format(name) + response = self.__session.request( + "POST", + self.__borrar_escenario_url, + data=delete_data, + headers=self.__headers, + ) if response.status_code != 200: raise ResponseException(response.status_code) return True def _consumption_raw(self, start, end): self.__check_session() - start_str = start.strftime('%d-%m-%Y') - end_str = end.strftime('%d-%m-%Y') + start_str = start.strftime("%d-%m-%Y") + end_str = end.strftime("%d-%m-%Y") - response = self.__session.request("GET", self.__obtener_periodo_url.format(start_str, end_str), headers=self.__headers) + response = self.__session.request( + "GET", + self.__obtener_periodo_url.format(start_str, end_str), + headers=self.__headers, + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -222,20 +293,23 @@ def _consumption_raw(self, start, end): def consumption(self, start, end): json = self._consumption_raw(start, end) values = [] - for x in json['y']['data'][0]: + for x in json["y"]["data"][0]: if x is None: values.append(None) else: - values.append(float(x['valor'])) + values.append(float(x["valor"])) return values def _production_raw(self, start, end): self.__check_session() - start_str = start.strftime('%d-%m-%Y') - end_str = end.strftime('%d-%m-%Y') + start_str = start.strftime("%d-%m-%Y") + end_str = end.strftime("%d-%m-%Y") - response = self.__session.request("GET", self.__obtener_periodo_generacion_url.format(start_str, end_str), - headers=self.__headers) + response = self.__session.request( + "GET", + self.__obtener_periodo_generacion_url.format(start_str, end_str), + headers=self.__headers, + ) if response.status_code != 200: raise ResponseException(response.status_code) if not response.text: @@ -253,15 +327,61 @@ def _production_raw(self, start, end): def production(self, start, end): json = self._production_raw(start, end) values = [] - for x in json['y']['data'][0]: + for x in json["y"]["data"][0]: if x is None: values.append(None) else: - values.append(float(x['valor'])) + values.append(float(x["valor"])) return values # Get total consumption in Wh (Watt-hour) over a time period # # start/end: datetime.date objects indicating the time period (both inclusive) def total_consumption(self, start, end): - return float(self._consumption_raw(start, end)['acumulado']) + return float(self._consumption_raw(start, end)["acumulado"]) + + def _consumption_facturado_raw(self, start, end): + self.__check_session() + start_str = start.strftime("%d-%m-%Y") + end_str = end.strftime("%d-%m-%Y") + + response = self.__session.request( + "GET", + self.__obtener_periodo_facturado_url.format(start_str, end_str), + headers=self.__headers, + ) + if response.status_code != 200: + raise ResponseException(response.status_code) + if not response.text: + raise NoResponseException + return response.json() + + # Get billed consumption data from a time period + # + # start/end: datetime.date objects indicating the time period (both inclusive) + # + # Returns a list of billed consumptions starting at midnight on the start day until 23:00 on the last day. + # Each value is the hourly billed consumption in Wh. + def billed_consumption(self, start, end): + json = self._consumption_facturado_raw(start, end) + values = [] + for x in json["y"]["data"][0]: + if x is None: + values.append(None) + else: + values.append(float(x["valor"])) + return values + + # Get total billed consumption in Wh (Watt-hour) over a time period + # + # start/end: datetime.date objects indicating the time period (both inclusive) + def total_billed_consumption(self, start, end): + return float(self._consumption_facturado_raw(start, end)["acumulado"]) + + @deprecated("Use 'billed_consumption' method instead") + def consumption_facturado(self, start, end): + return self.billed_consumption(start, end) + + @deprecated("Use 'total_billed_consumption' method instead") + def total_consumption_facturado(self, start, end): + return self.total_billed_consumption(start, end) diff --git a/setup.py b/setup.py index 9213c76..25b3fef 100644 --- a/setup.py +++ b/setup.py @@ -9,7 +9,7 @@ setuptools_git_versioning={ "enabled": True, }, - setup_requires=['setuptools-git-versioning'], + setup_requires=["setuptools-git-versioning"], author="hectorespert", author_email="hectorespertpardo@gmail.com", description="UNOFFICIAL Python client for i-DE", @@ -22,13 +22,8 @@ "Documentation": "https://github.com/hectorespert/python-oligo#readme", }, packages=find_packages(), - install_requires=[ - 'deprecated' - ], - extras_require={ - 'requests': ['requests'], - 'asyncio': ['aiohttp'] - }, + install_requires=["deprecated"], + extras_require={"requests": ["requests"], "asyncio": ["aiohttp"]}, classifiers=[ "Development Status :: 7 - Inactive", "Programming Language :: Python :: 3", @@ -36,5 +31,5 @@ "Operating System :: OS Independent", ], keywords=["i-DE", "iberdrola", "energy", "client"], - python_requires='>=3.6', + python_requires=">=3.6", ) diff --git a/tests/test_iber.py b/tests/test_iber.py index 5b61517..69fd20c 100644 --- a/tests/test_iber.py +++ b/tests/test_iber.py @@ -1,43 +1,124 @@ +import os import unittest +from datetime import date +from unittest.mock import MagicMock, patch from requests import Session -from requests_mock import Adapter from oligo import Iber from oligo.exception import LoginException, ResponseException class TestIber(unittest.TestCase): - def setUp(self): - self.adapter = Adapter() - + self._env_backup = {} + for key in ("I_DE_USER", "I_DE_PASSWORD"): + if key in os.environ: + self._env_backup[key] = os.environ.pop(key) self.session = Session() - self.session.mount('https://', self.adapter) - self.instance = Iber(self.session) + def tearDown(self): + for key, value in self._env_backup.items(): + os.environ[key] = value + def test_instance(self): self.assertIsInstance(self.instance, Iber) def test_login(self): - self.adapter.register_uri('POST', - 'https://www.i-de.es/consumidores/rest/loginNew/login', - text='{"success":"true"}') - self.instance.login("user", "password", self.session) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"success":"true"}' + mock_response.json.return_value = {"success": "true"} + + with patch.object(self.session, "request", return_value=mock_response): + self.instance.login("user", "password", self.session) + + def test_login_with_env_vars(self): + os.environ["I_DE_USER"] = "env_user" + os.environ["I_DE_PASSWORD"] = "env_pass" + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"success":"true"}' + mock_response.json.return_value = {"success": "true"} + + with patch.object( + self.session, "request", return_value=mock_response + ) as mock_request: + self.instance.login(session=self.session) + _, kwargs = mock_request.call_args + payload = kwargs["json"] + self.assertEqual(payload[0], "env_user") + self.assertEqual(payload[1], "env_pass") + + def test_login_explicit_args_override_env_vars(self): + os.environ["I_DE_USER"] = "env_user" + os.environ["I_DE_PASSWORD"] = "env_pass" + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"success":"true"}' + mock_response.json.return_value = {"success": "true"} + + with patch.object( + self.session, "request", return_value=mock_response + ) as mock_request: + self.instance.login("user", "password", self.session) + _, kwargs = mock_request.call_args + payload = kwargs["json"] + self.assertEqual(payload[0], "user") + self.assertEqual(payload[1], "password") def test_login_failed(self): - self.adapter.register_uri('POST', - 'https://www.i-de.es/consumidores/rest/loginNew/login', - text='{"success":"false"}') - self.assertRaises(LoginException, self.instance.login, "user", "password", self.session) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"success":"false"}' + mock_response.json.return_value = {"success": "false"} + + with patch.object(self.session, "request", return_value=mock_response): + self.assertRaises( + LoginException, self.instance.login, "user", "password", self.session + ) + + def test_login_missing_credentials(self): + self.assertRaises(LoginException, self.instance.login, session=self.session) def test_login_failed_by_status_code(self): - self.adapter.register_uri('POST', - 'https://www.i-de.es/consumidores/rest/loginNew/login', - status_code=500) - self.assertRaises(ResponseException, self.instance.login, "user", "password", self.session) + mock_response = MagicMock() + mock_response.status_code = 500 + + with patch.object(self.session, "request", return_value=mock_response): + self.assertRaises( + ResponseException, self.instance.login, "user", "password", self.session + ) + + def test_billed_consumption(self): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"y":{"data":[[{"valor":"1.5"},{"valor":"2.5"}]]}}' + mock_response.json.return_value = { + "y": {"data": [[{"valor": "1.5"}, {"valor": "2.5"}]]} + } + + with patch.object(self.session, "request", return_value=mock_response): + result = self.instance.billed_consumption( + date(2024, 1, 1), date(2024, 1, 2) + ) + self.assertEqual(result, [1.5, 2.5]) + + def test_total_billed_consumption(self): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"acumulado":"10.5"}' + mock_response.json.return_value = {"acumulado": "10.5"} + + with patch.object(self.session, "request", return_value=mock_response): + result = self.instance.total_billed_consumption( + date(2024, 1, 1), date(2024, 1, 2) + ) + self.assertEqual(result, 10.5) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main()