diff --git a/oligo/asyncio/asynciber.py b/oligo/asyncio/asynciber.py index dc6205a..ecd2d7c 100644 --- a/oligo/asyncio/asynciber.py +++ b/oligo/asyncio/asynciber.py @@ -1,3 +1,6 @@ +from oligo.exception import SessionException, ResponseException, NoResponseException, LoginException, \ + SelectContractException + try: import aiohttp except ImportError: @@ -20,26 +23,6 @@ OBTENER_PERIODO_GENERACION_URL = "consumoNew/obtenerDatosGeneracionPeriodo/fechaInicio/{}00:00:00/fechaFinal/{}00:00:00/" -class ResponseException(Exception): - pass - - -class LoginException(Exception): - pass - - -class SessionException(Exception): - pass - - -class NoResponseException(Exception): - pass - - -class SelectContractException(Exception): - pass - - class AsyncIber: def __init__(self) -> None: """Iber class __init__ method.""" @@ -52,9 +35,7 @@ async def __request( self, path: str, data: Optional[Union[list, dict]] = None ) -> dict: if not self.__session: - raise SessionException( - "Session required, use login() method to obtain a session" - ) + raise SessionException() if data is None: response = await self.__session.get( f"https://www.i-de.es/consumidores/rest/{path}", @@ -68,7 +49,7 @@ async def __request( ) if response.status != 200: self.__session = None - raise ResponseException("Response error, code: {}".format(response.status)) + raise ResponseException(response.status) data = await response.json() if not data: raise NoResponseException @@ -91,7 +72,7 @@ async def login(self, user: str, password: str) -> bool: data = await self.__request(LOGIN_URL, data=payload) if data["success"] != "true": self.__session = None - raise LoginException("Login error, bad login") + raise LoginException() return True async def measurement(self) -> dict: diff --git a/oligo/exception.py b/oligo/exception.py new file mode 100644 index 0000000..20fdf38 --- /dev/null +++ b/oligo/exception.py @@ -0,0 +1,25 @@ +class IberException(Exception): + pass + + +class ResponseException(IberException): + def __init__(self, status_code): + super().__init__("Response error, code: {}".format(status_code)) + + +class LoginException(IberException): + def __init__(self): + super().__init__('Login error, bad login') + + +class SessionException(IberException): + def __init__(self): + super().__init__('Session required, use login() method to obtain a session') + + +class NoResponseException(IberException): + pass + + +class SelectContractException(IberException): + pass diff --git a/oligo/iber.py b/oligo/iber.py index 5948eb6..ba6e9c0 100644 --- a/oligo/iber.py +++ b/oligo/iber.py @@ -1,25 +1,8 @@ -from requests import Session from datetime import datetime +from requests import Session -class ResponseException(Exception): - pass - - -class LoginException(Exception): - pass - - -class SessionException(Exception): - pass - - -class NoResponseException(Exception): - pass - - -class SelectContractException(Exception): - pass +from .exception import LoginException, ResponseException, NoResponseException, SelectContractException, SessionException class Iber: @@ -56,22 +39,22 @@ def login(self, user, password, session=Session()): response = self.__session.request("POST", self.__login_url, data=login_data, headers=self.__headers) if response.status_code != 200: self.__session = None - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) json_response = response.json() if json_response["success"] != "true": self.__session = None - raise LoginException("Login error, bad login") + raise LoginException() def __check_session(self): if not self.__session: - raise SessionException("Session required, use login() method to obtain a session") + raise SessionException() def measurement(self): """Returns a measurement from the powermeter.""" self.__check_session() response = self.__session.request("GET", self.__watthourmeter_url, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -91,7 +74,7 @@ def icpstatus(self): self.__check_session() response = self.__session.request("POST", self.__icp_status_url, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -104,7 +87,7 @@ def contracts(self): self.__check_session() response = self.__session.request("GET", self.__contracts_url, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -115,7 +98,7 @@ def contract(self): self.__check_session() response = self.__session.request("GET", self.__contract_detail_url, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException return response.json() @@ -124,7 +107,7 @@ def contractselect(self, id): self.__check_session() response = self.__session.request("GET", self.__contract_selection_url + id, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -135,7 +118,7 @@ def scene_list(self): self.__check_session() response = self.__session.request("GET", self.__obtener_escenarios_url, headers=self.__headers) if response.status_code != 200: - raise ResponseException("Response error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -149,7 +132,7 @@ def scene_get(self, name): 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 error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -167,7 +150,7 @@ def scene_save(self, consumption, measurement_id, description): 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 error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException json_response = response.json() @@ -181,7 +164,7 @@ def scene_delete(self, name): 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 error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) return True def _consumption_raw(self, start, end): @@ -191,7 +174,7 @@ def _consumption_raw(self, start, end): 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 error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException return response.json() @@ -222,7 +205,7 @@ def _production_raw(self, start, end): 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 error, code: {}".format(response.status_code)) + raise ResponseException(response.status_code) if not response.text: raise NoResponseException return response.json() diff --git a/tests/test_exception.py b/tests/test_exception.py new file mode 100644 index 0000000..9c1cd41 --- /dev/null +++ b/tests/test_exception.py @@ -0,0 +1,24 @@ +import unittest + +from oligo.exception import SessionException, LoginException, ResponseException + + +class TestResponseException(unittest.TestCase): + + def test_message(self): + login_exception = ResponseException(418) + self.assertEqual('Response error, code: 418', login_exception.args[0]) + + +class TestLoginException(unittest.TestCase): + + def test_message(self): + login_exception = LoginException() + self.assertEqual('Login error, bad login', login_exception.args[0]) + + +class TestSessionException(unittest.TestCase): + + def test_message(self): + session_exception = SessionException() + self.assertEqual('Session required, use login() method to obtain a session', session_exception.args[0]) diff --git a/tests/test_iber.py b/tests/test_iber.py index d079829..5b61517 100644 --- a/tests/test_iber.py +++ b/tests/test_iber.py @@ -4,7 +4,7 @@ from requests_mock import Adapter from oligo import Iber -from oligo.iber import LoginException, ResponseException +from oligo.exception import LoginException, ResponseException class TestIber(unittest.TestCase):