Merge pull request #12 from adator85/dev

Dev

Author: adator85

dist/unrealircd_rpc_py-0.1.2-py3-none-any.whl

@@ -2,7 +2,7 @@
 
 setup(
     name='unrealircd_rpc_py',
-    version='0.1.1',
+    version='0.1.4',
     packages=find_packages(),
     install_requires=[
         "requests>=2.25.1"

dist/unrealircd_rpc_py-0.1.2.tar.gz

@@ -1,5 +1,5 @@
 import json.scanner
-import requests, json, urllib3, socket, re, sys
+import requests, json, urllib3, socket, re, sys, os
 from requests.auth import HTTPBasicAuth
 import base64, ssl, time, logging, random
 from typing import Literal, Union
@@ -18,6 +18,7 @@ def __init__(self, req_method:str, url: str, path_to_socket_file: str, username:
         self.debug_level = debug_level
         self.Logs: logging
         self.__init_log_system()
+        self.Error = self.ErrorModel(0, '')
 
         self.url = url
         self.path_to_socket_file = path_to_socket_file
@@ -28,19 +29,14 @@ def __init__(self, req_method:str, url: str, path_to_socket_file: str, username:
         self.username = username
         self.password = password
 
-        if not self.__check_url(url) and not url is None:
-            self.Logs.critical('You must provide the url in this format: https://your.rpcjson.link:port/api')
-            sys.exit(3)
-
         self.request: str = ''
         self.req_method = req_method
         self.str_response = ''
         self.json_response = ''
 
-        # Option 2 with Namespaces
+        # Option 2 with Namespacescs
         self.json_response_np: SimpleNamespace
-
-        self.Error = self.ErrorModel(0, '')
+        self.query('stats.get')
 
     def __check_url(self, url: str) -> bool:
         """Check provided url if it follow the format
@@ -66,16 +62,66 @@ def __check_url(self, url: str) -> bool:
                 self.port = match.group(2)
                 self.endpoint = match.group(3)
                 response = True
+            else:
+                self.Error.code = -1
+                self.Error.message = 'You must provide the url in this format: https://your.rpcjson.link:port/api'
+
+            return response
+        except NameError as nameerr:
+            self.Logs.critical(f'NameError: {nameerr}')
+
+    def __check_unix_socket_file(self, path_to_socket_file: str) -> bool:
+        """Check provided full path to socket file if it exist
+
+        Args:
+            path_to_socket_file (str): Full path to unix socket file
+
+        Returns:
+            bool: True if path is correct else False
+        """
+        try:
+            response = False
+
+            if path_to_socket_file is None:
+                self.Error.code = -1
+                self.Error.message = 'The Path to your socket file is empty ? please be sure that you are providing the correct socket path'
+                return response
+
+            if not os.path.exists(path_to_socket_file):
+                self.Error.code = -1
+                self.Error.message = 'The Path to your socket file is wrong ? please make sure that you are providing the correct socket path'
+                return response
+
+            response = True
 
             return response
         except NameError as nameerr:
             self.Logs.critical(f'NameError: {nameerr}')
 
+    def __is_error_connection(self, response: str) -> bool:
+        """If True, it means that there is an error
+
+        Args:
+            response (str): The response to analyse
+
+        Returns:
+            bool: True if there is a connection error
+        """
+        if 'authentication required' == response.lower().strip():
+            self.Error.code = -1
+            self.Error.message = '>> Authentication required'
+            return True
+        else:
+            return False
+
     def __send_to_unixsocket(self):
         try:
 
             sock = socket.socket(socket.AddressFamily.AF_UNIX, socket.SocketKind.SOCK_STREAM)
 
+            if not self.__check_unix_socket_file(self.path_to_socket_file):
+                return None
+
             sock.connect(self.path_to_socket_file)
             sock.settimeout(10)
 
@@ -102,16 +148,21 @@ def __send_to_unixsocket(self):
 
         except AttributeError as attrerr:
             self.Logs.critical(f'AF_Unix Error: {attrerr}')
-            sys.exit('AF_UNIX Are you sure you want to use Unix socket ?')
+            self.Error.code = -1
+            self.Error.message = 'AF_UNIX Are you sure you want to use Unix socket ?'
         except OSError as oserr:
             self.Logs.critical(f'System Error: {oserr}')
-            sys.exit(3)
         except Exception as err:
             self.Logs.error(f'General Error: {err}')
 
     def __send_srequest(self):
         """S For socket connection"""
         try:
+
+            if not self.__check_url(self.url) and not self.url is None:
+                self.Logs.critical('You must provide the url in this format: https://your.rpcjson.link:port/api')
+                return None
+
             get_url = self.url
             get_host = self.host
             get_port = self.port
@@ -148,15 +199,26 @@ def __send_srequest(self):
                     else:
                         body = response_str
 
+                    if self.__is_error_connection(body):
+                        return None
+
                     self.json_response = json.loads(body)
                     self.json_response_np: SimpleNamespace = json.loads(body, object_hook=lambda d: SimpleNamespace(**d))
 
+        except (socket.error, ssl.SSLError) as serror:
+            self.Logs.error(f'Socket Error: {serror}')
         except Exception as err:
             self.Logs.error(f'General Error: {err}')
+            # self.Logs.error(f'General Error: {traceback.format_exc()}')
 
     def __send_request(self) :
         """Use requests module"""
         try:
+
+            if not self.__check_url(self.url) and not self.url is None:
+                self.Logs.critical('You must provide the url in this format: https://your.rpcjson.link:port/api')
+                return None
+
             verify = False
 
             urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@@ -166,6 +228,9 @@ def __send_request(self) :
 
             response = requests.post(url=self.url, auth=credentials, data=jsonrequest, verify=verify)
 
+            if self.__is_error_connection(response.text):
+                return None
+
             decodedResponse = json.dumps(response.text)
 
             self.str_response = decodedResponse
@@ -181,7 +246,6 @@ def __send_request(self) :
         except requests.ConnectionError as ce:
             self.Logs.critical(f"Connection Error : {ce}")
             self.Logs.critical(f"Initial request: {self.request}")
-            sys.exit(3)
         except json.decoder.JSONDecodeError as jsonerror:
             self.Logs.error(f"jsonError {jsonerror}")
             self.Logs.error(f"Initial request: {self.request}")

dist/unrealircd_rpc_py-0.1.3-py3-none-any.whl

@@ -18,9 +18,13 @@ def __init__(self, path_to_socket_file: str, callback_object_instance: object, c
         self.Logs: logging
         self.__init_log_system()
 
+        self.Error = self.ErrorModel(0, '')
+
         if not self.__check_unix_socket_file(path_to_socket_file=path_to_socket_file):
             self.Logs.critical(f'The socket file is not available, please check the full path of your socket file')
-            sys.exit('please check the full path of your socket file')
+            self.Error.code = -1
+            self.Error.message = 'The socket file is not available, please check the full path of your socket file'
+            return None
 
         self.to_run = getattr(callback_object_instance, callback_method_name)
 
@@ -35,8 +39,6 @@ def __init__(self, path_to_socket_file: str, callback_object_instance: object, c
         # Option 2 with Namespaces
         self.json_response_np: SimpleNamespace
 
-        self.Error = self.ErrorModel(0, '')
-
     def __check_unix_socket_file(self, path_to_socket_file: str) -> bool:
         """Check provided full path to socket file if it exist
 

dist/unrealircd_rpc_py-0.1.3.tar.gz

@@ -143,9 +143,9 @@ def get(self, nickoruid: str) -> Union[ModelUser, None]:
                     hostname=user['hostname'] if 'hostname' in user else None,
                     ip=user['ip'] if 'ip' in user else None,
                     details=user['details'] if 'details' in user else None,
-                    server_port=user['server_port'] if 'server_port' in user  else 0,
-                    client_port=user['client_port'] if 'client_port' in user  else 0,
-                    connected_since=user['connected_since'] if 'connected_since' in user  else None,
+                    server_port=user['server_port'] if 'server_port' in user else 0,
+                    client_port=user['client_port'] if 'client_port' in user else 0,
+                    connected_since=user['connected_since'] if 'connected_since' in user else None,
                     idle_since=user['idle_since'] if 'idle_since' in user  else None,
                     username=user['user']['username'] if 'user' in user and 'username' in user['user'] else None,
                     realname=user['user']['realname'] if 'user' in user and 'realname' in user['user'] else None,
@@ -155,7 +155,7 @@ def get(self, nickoruid: str) -> Union[ModelUser, None]:
                     reputation=user['user']['reputation'] if 'user' in user and 'reputation' in user['user'] else 0,
                     security_groups=user['user']['security-groups'] if 'user' in user and 'security-groups' in user['user'] else [],
                     modes=user['user']['modes'] if 'user' in user and 'modes' in user['user'] else None,
-                    channels=user['user']['channels'],
+                    channels=user['user']['channels'] if 'user' in user and 'channels' in user['user'] else [],
                     cipher=user['tls']['cipher'] if 'tls' in user and 'cipher' in user['tls'] else None,
                     certfp=user['tls']['certfp'] if 'tls' in user and 'certfp' in user['tls'] else None,
                     country_code=user['geoip']['country_code'] if 'geoip' in user and 'country_code' in user['geoip'] else None,