Use common exceptions

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:

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

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()

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])

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):