From 4164c3592efacfb954031dba3db0aa8673363178 Mon Sep 17 00:00:00 2001 From: Oscar Neira Date: Sun, 24 May 2026 08:11:36 +0200 Subject: [PATCH 1/2] Added consumo:_historico and python-dotenv for reading username and password from env vars --- README.md | 116 ++++++++++++++++++++ oligo/asyncio/asynciber.py | 52 ++++++++- oligo/exception.py | 8 +- oligo/requests/iber.py | 212 +++++++++++++++++++++++++++---------- requirements.txt | 1 + setup.py | 13 +-- tests/test_iber.py | 104 ++++++++++++++---- 7 files changed, 418 insertions(+), 88 deletions(-) diff --git a/README.md b/README.md index 9398143..e824ca3 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` (también se admiten ficheros `.env` gracias a `python-dotenv`): + +```python +from oligo import Iber + +connection = Iber() +connection.login("user", "password") +``` + +O usando variables de entorno / `.env`: + +```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.consumption_facturado(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.consumption_facturado(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 (`.env` files are also supported via `python-dotenv`): + +```python +from oligo import Iber + +connection = Iber() +connection.login("user", "password") +``` + +Or using environment variables / `.env`: + +```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.consumption_facturado(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.consumption_facturado(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..3676fee 100644 --- a/oligo/asyncio/asynciber.py +++ b/oligo/asyncio/asynciber.py @@ -1,7 +1,15 @@ +import os + +from dotenv import load_dotenv from deprecated.classic import deprecated -from ..exception import SessionException, ResponseException, NoResponseException, LoginException, \ - SelectContractException +from ..exception import ( + SessionException, + ResponseException, + NoResponseException, + LoginException, + SelectContractException, +) try: import aiohttp @@ -11,6 +19,8 @@ from datetime import datetime from typing import Union, Optional +load_dotenv() + LOGIN_URL = "loginNew/login" WATTHOURMETER_URL = "escenarioNew/obtenerMedicionOnline/24" ICP_STATUS_URL = "rearmeICP/consultarEstado" @@ -23,6 +33,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,9 +70,18 @@ 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: str = None, password: str = None) -> bool: + """Creates session with your credentials. + Reads I-DE-USER and I-DE-PASSWORD from environment if available, + falling back to the provided parameters.""" self.__session = aiohttp.ClientSession() + user = os.getenv("I-DE-USER", user) + password = os.getenv("I-DE-PASSWORD", 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.", + ) payload = [ user, password, @@ -197,3 +217,27 @@ 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 consumption_facturado(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_consumption_facturado(self, start, end) -> float: + data = await self._consumption_facturado_raw(start, end) + return float(data["acumulado"]) 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..5b2f3ef 100644 --- a/oligo/requests/iber.py +++ b/oligo/requests/iber.py @@ -1,5 +1,7 @@ +import os from datetime import datetime +from dotenv import load_dotenv from deprecated.classic import deprecated try: @@ -7,25 +9,51 @@ except ImportError: raise RuntimeError("Iber requires requests module") -from ..exception import LoginException, ResponseException, NoResponseException, SelectContractException, \ - SessionException +load_dotenv() +from ..exception import ( + LoginException, + ResponseException, + NoResponseException, + SelectContractException, + SessionException, +) -class Iber: +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 +71,18 @@ 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=Session()): + """Creates session with your credentials. + Reads I-DE-USER and I-DE-PASSWORD from environment if available, + falling back to the provided parameters.""" self.__session = session + user = os.getenv("I-DE-USER", user) + password = os.getenv("I-DE-PASSWORD", 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 +95,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 +113,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 +135,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 +145,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 +160,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 +173,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 +184,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 +197,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 +287,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 +321,53 @@ 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 consumption_facturado(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_consumption_facturado(self, start, end): + return float(self._consumption_facturado_raw(start, end)["acumulado"]) diff --git a/requirements.txt b/requirements.txt index a04cd47..760e91d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ requests requests-mock aiohttp deprecated +python-dotenv diff --git a/setup.py b/setup.py index 9213c76..f98b1ce 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", "python-dotenv"], + 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..efcab28 100644 --- a/tests/test_iber.py +++ b/tests/test_iber.py @@ -1,43 +1,109 @@ +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("user", "password", self.session) + _, kwargs = mock_request.call_args + payload = kwargs["json"] + self.assertEqual(payload[0], "env_user") + self.assertEqual(payload[1], "env_pass") + + del os.environ["I-DE-USER"] + del os.environ["I-DE-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_consumption_facturado(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.consumption_facturado( + date(2024, 1, 1), date(2024, 1, 2) + ) + self.assertEqual(result, [1.5, 2.5]) + + def test_total_consumption_facturado(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_consumption_facturado( + date(2024, 1, 1), date(2024, 1, 2) + ) + self.assertEqual(result, 10.5) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main() From 08dd840882177af27e4d9dc25fc0ebab9a8666dc Mon Sep 17 00:00:00 2001 From: Oscar Neira Date: Tue, 26 May 2026 04:41:52 +0200 Subject: [PATCH 2/2] user and passwrod from commandlie have preference over env vars --- oligo/requests/iber.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/oligo/requests/iber.py b/oligo/requests/iber.py index 5b2f3ef..dc564a9 100644 --- a/oligo/requests/iber.py +++ b/oligo/requests/iber.py @@ -76,8 +76,8 @@ def login(self, user=None, password=None, session=Session()): Reads I-DE-USER and I-DE-PASSWORD from environment if available, falling back to the provided parameters.""" self.__session = session - user = os.getenv("I-DE-USER", user) - password = os.getenv("I-DE-PASSWORD", password) + user = user or os.getenv("I-DE-USER", user) + password = password or os.getenv("I-DE-PASSWORD", password) if not user or not password: raise LoginException( user or "unknown",