Add status and product info support and Issue #4

Author: rlabbeptc

kepconfig/connection.py

@@ -152,7 +152,37 @@ def SSL_trust_all_certs(self, val):
                 self.__ssl_context.verify_mode = ssl.CERT_REQUIRED
 
 
+    def get_status(self) -> dict:
+        '''Executes a health status request to the Kepware instance to report service statuses.
 
+        RETURNS:
+
+        list - list of data for the health status request
+
+        EXCEPTIONS:
+        
+        KepHTTPError - If urllib provides an HTTPError
+        KepURLError - If urllib provides an URLError
+        '''
+
+        r = self._config_get(f'{self.url}/status')
+        return r.payload
+    def get_info(self) -> dict:
+        '''Requests product information from the Kepware instance. Provides various information including
+        product name and version information.
+
+        RETURNS:
+
+        dict - data for the product information request
+
+        EXCEPTIONS:
+        
+        KepHTTPError - If urllib provides an HTTPError
+        KepURLError - If urllib provides an URLError
+        '''
+        
+        r = self._config_get(f'{self.url}/about')
+        return r.payload
 
     def reinitialize(self, job_ttl = None) -> KepServiceResponse:
         '''Executes a Reinitialize call to the Kepware instance.
@@ -337,6 +367,9 @@ def _config_add(self, url, DATA):
         '''Conducts an POST method at *url* to add an object in the Kepware Configuration
         *DATA* is required to be a properly JSON object (dict) of the item to be posted to *url* 
         '''
+        if len(DATA) == 0:
+            err_msg = f'Error: Empty List or Dict in DATA | DATA type: {type(DATA)}'
+            raise KepError(err_msg) 
         data = json.dumps(DATA).encode('utf-8')
         url_obj = self.__url_validate(url)
         q = request.Request(url_obj, data, method='POST')

tests/connectivity_test.py

@@ -77,6 +77,20 @@ def test_channel_add(server):
     channel_data = {"common.ALLTYPES_NAME": ch_name,"servermain.MULTIPLE_TYPES_DEVICE_DRIVER": "Simulator"}
     assert kepconfig.connectivity.channel.add_channel(server,channel_data)
 
+    # Add a empty Channel objects to validate error handling. Validates low level _config_add() methond in Server class
+    # used by all add methods across the package
+    channel_data = {}
+    # pytest.raises(kepconfig.error.KepError,kepconfig.connectivity.channel.add_channel, {'server': server, 'DATA': channel_data})
+    with pytest.raises(kepconfig.error.KepError) as e:
+        kepconfig.connectivity.channel.add_channel(server,channel_data)
+    assert e.type == kepconfig.error.KepError
+    # assert type(kepconfig.connectivity.channel.add_channel(server,channel_data)) == kepconfig.error.KepError
+
+    channel_data = []
+    with pytest.raises(kepconfig.error.KepError) as e:
+        kepconfig.connectivity.channel.add_channel(server,channel_data)
+    assert e.type == kepconfig.error.KepError
+
     # Add multi channels with one failure
     channel_data = [
         {"common.ALLTYPES_NAME": ch_name+"1","servermain.MULTIPLE_TYPES_DEVICE_DRIVER": "Simulator"},