fix #272 (#360)

* fix #272

* fix wrong Errortype

Author: swamper123

snap7/client.py

@@ -10,7 +10,6 @@
 
 import snap7
 from snap7.common import check_error, ipv4, load_library
-from snap7.exceptions import Snap7Exception
 from snap7.types import S7SZL, Areas, BlocksList, S7CpInfo, S7CpuInfo, S7DataItem
 from snap7.types import S7OrderCode, S7Protection, S7SZLList, TS7BlockInfo, WordLen
 from snap7.types import S7Object, buffer_size, buffer_type, cpu_statuses, param_types
@@ -125,7 +124,7 @@ def get_cpu_state(self) -> str:
             Description of the cpu state.
 
         Raises:
-            :obj:`Snap7Exception`: if the cpu state is invalid.
+            :obj:`ValueError`: if the cpu state is invalid.
 
         Examples:
             >>> client.get_cpu_statE()
@@ -136,7 +135,7 @@ def get_cpu_state(self) -> str:
         try:
             status_string = cpu_statuses[state.value]
         except KeyError:
-            raise Snap7Exception(f"The cpu state ({state.value}) is invalid")
+            raise ValueError(f"The cpu state ({state.value}) is invalid")
 
         logger.debug(f"CPU state is {status_string}")
         return status_string
@@ -481,12 +480,12 @@ def list_blocks_of_type(self, blocktype: str, size: int) -> Union[int, Array]:
             If size is 0, it returns a 0, otherwise an `Array` of specified block type.
 
         Raises:
-            :obj:`Snap7Exception`: if the `blocktype` is not valid.
+            :obj:`ValueError`: if the `blocktype` is not valid.
         """
 
         _blocktype = snap7.types.block_types.get(blocktype)
         if not _blocktype:
-            raise Snap7Exception("The blocktype parameter was invalid")
+            raise ValueError("The blocktype parameter was invalid")
 
         logger.debug(f"listing blocks of type: {_blocktype} size: {size}")
 
@@ -516,7 +515,7 @@ def get_block_info(self, blocktype: str, db_number: int) -> TS7BlockInfo:
             Structure of information from block.
 
         Raises:
-            :obj:`Snap7Exception`: if the `blocktype` is not valid.
+            :obj:`ValueError`: if the `blocktype` is not valid.
 
         Examples:
             >>> block_info = client.get_block_info("DB", 1)
@@ -540,7 +539,7 @@ def get_block_info(self, blocktype: str, db_number: int) -> TS7BlockInfo:
         blocktype_ = snap7.types.block_types.get(blocktype)
 
         if not blocktype_:
-            raise Snap7Exception("The blocktype parameter was invalid")
+            raise ValueError("The blocktype parameter was invalid")
         logger.debug(f"retrieving block info for block {db_number} of type {blocktype_}")
 
         data = TS7BlockInfo()
@@ -589,7 +588,7 @@ def set_connection_params(self, address: str, local_tsap: int, remote_tsap: int)
 
         Raises:
             :obj:`ValueError`: if the `address` is not a valid IPV4.
-            :obj:`Snap7Exception`: if the result of setting the connection params is
+            :obj:`ValueError`: if the result of setting the connection params is
                 different than 0.
         """
         if not re.match(ipv4, address):
@@ -598,7 +597,7 @@ def set_connection_params(self, address: str, local_tsap: int, remote_tsap: int)
                                                        c_uint16(local_tsap),
                                                        c_uint16(remote_tsap))
         if result != 0:
-            raise Snap7Exception("The parameter was invalid")
+            raise ValueError("The parameter was invalid")
 
     def set_connection_type(self, connection_type: int):
         """ Sets the connection resource type, i.e the way in which the Clients connects to a PLC.
@@ -607,13 +606,13 @@ def set_connection_type(self, connection_type: int):
             connection_type: 1 for PG, 2 for OP, 3 to 10 for S7 Basic
 
         Raises:
-            :obj:`Snap7Exception`: if the result of setting the connection type is
+            :obj:`ValueError`: if the result of setting the connection type is
                 different than 0.
         """
         result = self._library.Cli_SetConnectionType(self._pointer,
                                                      c_uint16(connection_type))
         if result != 0:
-            raise Snap7Exception("The parameter was invalid")
+            raise ValueError("The parameter was invalid")
 
     def get_connected(self) -> bool:
         """Returns the connection status
@@ -1122,11 +1121,11 @@ def as_list_blocks_of_type(self, blocktype: str, data, count) -> int:
             Snap7 code.
 
         Raises:
-            :obj:`Snap7Exception`: if the `blocktype` is invalid
+            :obj:`ValueError`: if the `blocktype` is invalid
         """
         _blocktype = snap7.types.block_types.get(blocktype)
         if not _blocktype:
-            raise Snap7Exception("The blocktype parameter was invalid")
+            raise ValueError("The blocktype parameter was invalid")
         result = self._library.Cli_AsListBlocksOfType(self._pointer, _blocktype, byref(data), byref(count))
         check_error(result, context="client")
         return result

snap7/common.py

@@ -7,8 +7,6 @@
 from typing import Optional
 from ctypes.util import find_library
 
-from snap7.exceptions import Snap7Exception
-
 if platform.system() == 'Windows':
     from ctypes import windll as cdll  # type: ignore
 else:
@@ -52,7 +50,7 @@ def __init__(self, lib_location: Optional[str] = None):
             lib_location: full path to the `snap7.dll` file. Optional.
 
         Raises:
-            Snap7Exception: if `lib_location` is not found.
+            RuntimeError: if `lib_location` is not found.
         """
         if self.cdll:  # type: ignore
             return
@@ -62,8 +60,7 @@ def __init__(self, lib_location: Optional[str] = None):
                              or find_library('snap7')
                              or find_locally('snap7'))
         if not self.lib_location:
-            msg = "can't find snap7 library. If installed, try running ldconfig"
-            raise Snap7Exception(msg)
+            raise RuntimeError("can't find snap7 library. If installed, try running ldconfig")
         self.cdll = cdll.LoadLibrary(self.lib_location)
 
 
@@ -84,12 +81,12 @@ def check_error(code: int, context: str = "client") -> None:
         context: context in which is called.
 
     Raises:
-        Snap7Exception: if the code exists and is diferent from 1.
+        RuntimeError: if the code exists and is different from 1.
     """
     if code and code != 1:
         error = error_text(code, context)
         logger.error(error)
-        raise Snap7Exception(error)
+        raise RuntimeError(error)
 
 
 def error_text(error, context: str = "client") -> bytes:

snap7/logo.py

@@ -10,7 +10,6 @@
 from snap7 import types
 from snap7.types import WordLen, S7Object, param_types
 from snap7.common import ipv4, check_error, load_library
-from snap7.exceptions import Snap7Exception
 
 logger = logging.getLogger(__name__)
 
@@ -268,15 +267,15 @@ def set_connection_params(self, ip_address: str, tsap_snap7: int, tsap_logo: int
 
         Raises:
             :obj:`ValueError`: if the `ip_address` is not an IPV4.
-            :obj:`Snap7Exception`: if the snap7 error code is diferent from 0.
+            :obj:`ValueError`: if the snap7 error code is diferent from 0.
         """
         if not re.match(ipv4, ip_address):
             raise ValueError(f"{ip_address} is invalid ipv4")
         result = self.library.Cli_SetConnectionParams(self.pointer, ip_address.encode(),
                                                       c_uint16(tsap_snap7),
                                                       c_uint16(tsap_logo))
         if result != 0:
-            raise Snap7Exception("The parameter was invalid")
+            raise ValueError("The parameter was invalid")
 
     def set_connection_type(self, connection_type: int):
         """Sets the connection resource type, i.e the way in which the Clients
@@ -286,12 +285,12 @@ def set_connection_type(self, connection_type: int):
             connection_type: 1 for PG, 2 for OP, 3 to 10 for S7 Basic
 
         Raises:
-            :obj:`Snap7Exception`: if the snap7 error code is diferent from 0.
+            :obj:`ValueError`: if the snap7 error code is diferent from 0.
         """
         result = self.library.Cli_SetConnectionType(self.pointer,
                                                     c_uint16(connection_type))
         if result != 0:
-            raise Snap7Exception("The parameter was invalid")
+            raise ValueError("The parameter was invalid")
 
     def get_connected(self) -> bool:
         """Returns the connection status

snap7/partner.py

@@ -14,7 +14,6 @@
 
 import snap7.types
 from snap7.common import ipv4, check_error, load_library
-from snap7.exceptions import Snap7Exception
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +85,7 @@ def check_as_b_send_completion(self) -> Tuple[str, c_int32]:
         }
 
         if result == -2:
-            raise Snap7Exception("The Client parameter was invalid")
+            raise ValueError("The Client parameter was invalid")
 
         return return_values[result], op_result
 

snap7/util.py

@@ -92,7 +92,6 @@
 
 from snap7.types import Areas
 from snap7.client import Client
-from snap7.exceptions import Snap7Exception
 
 logger = logging.getLogger(__name__)
 
@@ -1167,7 +1166,7 @@ def get_value(self, byte_index: Union[str, int], type_: str) -> Union[ValueError
             type_: type of data to read.
 
         Raises:
-            :obj:`Snap7Exception`: if reading a `string` when checking the lenght of the string.
+            :obj:`ValueError`: if reading a `string` when checking the lenght of the string.
             :obj:`ValueError`: if the `type_` is not handled.
 
         Returns:
@@ -1190,7 +1189,7 @@ def get_value(self, byte_index: Union[str, int], type_: str) -> Union[ValueError
         if type_.startswith('STRING'):
             max_size = re.search(r'\d+', type_)
             if max_size is None:
-                raise Snap7Exception("Max size could not be determinate. re.search() returned None")
+                raise ValueError("Max size could not be determinate. re.search() returned None")
             max_size_grouped = max_size.group(0)
             max_size_int = int(max_size_grouped)
             return get_string(bytearray_, byte_index, max_size_int)
@@ -1226,7 +1225,7 @@ def set_value(self, byte_index: Union[str, int], type: str, value: Union[bool, s
             value: value to write.
 
         Raises:
-            :obj:`Snap7Exception`: if reading a `string` when checking the length of the string.
+            :obj:`ValueError`: if reading a `string` when checking the length of the string.
             :obj:`ValueError`: if the `type_` is not handled.
 
         Returns:
@@ -1245,7 +1244,7 @@ def set_value(self, byte_index: Union[str, int], type: str, value: Union[bool, s
         if type.startswith('STRING') and isinstance(value, str):
             max_size = re.search(r'\d+', type)
             if max_size is None:
-                raise Snap7Exception("Max size could not be determinate. re.search() returned None")
+                raise ValueError("Max size could not be determinate. re.search() returned None")
             max_size_grouped = max_size.group(0)
             max_size_int = int(max_size_grouped)
             return set_string(bytearray_, byte_index, value, max_size_int)

test/test_client.py

@@ -12,7 +12,6 @@
 
 import snap7
 from snap7 import util
-from snap7.exceptions import Snap7Exception
 from snap7.common import check_error
 from snap7.server import mainloop
 from snap7.types import S7AreaDB, S7DataItem, S7SZL, S7SZLList, buffer_type, buffer_size, S7Object, Areas, WordLen
@@ -61,7 +60,7 @@ def _as_check_loop(self, check_times=20) -> int:
                 break
             time.sleep(0.5)
         else:
-            raise Snap7Exception(f"Async Request not finished after {check_times} times - Fail")
+            raise TimeoutError(f"Async Request not finished after {check_times} times - Fail")
         return check_status.value
 
     def test_db_read(self):
@@ -156,13 +155,13 @@ def test_upload(self):
         this raises an exception due to missing authorization? maybe not
         implemented in server emulator
         """
-        self.assertRaises(Snap7Exception, self.client.upload, db_number)
+        self.assertRaises(RuntimeError, self.client.upload, db_number)
 
     def test_as_upload(self):
         _buffer = buffer_type()
         size = ctypes.c_int(ctypes.sizeof(_buffer))
         self.client.as_upload(1, _buffer, size)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     @unittest.skip("TODO: invalid block size")
     def test_download(self):
@@ -229,7 +228,7 @@ def test_list_blocks(self):
     def test_list_blocks_of_type(self):
         self.client.list_blocks_of_type('DB', 10)
 
-        self.assertRaises(Snap7Exception, self.client.list_blocks_of_type, 'NOblocktype', 10)
+        self.assertRaises(ValueError, self.client.list_blocks_of_type, 'NOblocktype', 10)
 
     def test_get_block_info(self):
         """test Cli_GetAgBlockInfo"""
@@ -592,7 +591,7 @@ def test_wait_as_completion_timeouted(self, timeout=0, tries=500):
             try:
                 res = self.client.wait_as_completion(timeout)
                 check_error(res)
-            except Snap7Exception as s7_err:
+            except RuntimeError as s7_err:
                 if not s7_err.args[0] == b'CLI : Job Timeout':
                     self.fail(f"While waiting another error appeared: {s7_err}")
                 # Wait for a thread to finish
@@ -712,24 +711,24 @@ def test_as_eb_read(self):
         buffer = (type_ * 1)()
         response = self.client.as_eb_read(0, 1, buffer)
         self.assertEqual(0, response)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_eb_write(self):
         # Cli_AsEBWrite
         response = self.client.as_eb_write(0, 1, bytearray(b'\x00'))
         self.assertEqual(0, response)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_full_upload(self):
         # Cli_AsFullUpload
         self.client.as_full_upload('DB', 1)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_list_blocks_of_type(self):
         data = (ctypes.c_uint16 * 10)()
         count = ctypes.c_int()
         self.client.as_list_blocks_of_type('DB', data, count)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_mb_read(self):
         # Cli_AsMBRead
@@ -738,13 +737,13 @@ def test_as_mb_read(self):
         data = (type_ * 1)()
         self.client.as_mb_read(0, 1, data)
         bytearray(data)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_mb_write(self):
         # Cli_AsMBWrite
         response = self.client.as_mb_write(0, 1, bytearray(b'\x00'))
         self.assertEqual(0, response)
-        self.assertRaises(Snap7Exception, self.client.wait_as_completion, 500)
+        self.assertRaises(RuntimeError, self.client.wait_as_completion, 500)
 
     def test_as_read_szl(self):
         # Cli_AsReadSZL
@@ -927,8 +926,8 @@ def test_read_szl(self):
         # read_szl_invalid_id
         ssl_id = 0xffff
         index = 0xffff
-        self.assertRaises(Snap7Exception, self.client.read_szl, ssl_id)
-        self.assertRaises(Snap7Exception, self.client.read_szl, ssl_id, index)
+        self.assertRaises(RuntimeError, self.client.read_szl, ssl_id)
+        self.assertRaises(RuntimeError, self.client.read_szl, ssl_id, index)
 
     def test_read_szl_list(self):
         # Cli_ReadSZLList
@@ -952,7 +951,7 @@ def test_tm_write(self):
         data = b'\x10\x01'
         self.assertEqual(0, self.client.tm_write(0, 1, bytearray(data)))
         self.assertEqual(data, self.client.tm_read(0, 1))
-        self.assertRaises(Snap7Exception, self.client.tm_write, 0, 100, bytes(200))
+        self.assertRaises(RuntimeError, self.client.tm_write, 0, 100, bytes(200))
         self.assertRaises(ValueError, self.client.tm_write, 0, 2, bytes(2))
 
     def test_write_multi_vars(self):

test/test_partner.py

@@ -113,7 +113,7 @@ def test_stop(self):
         self.partner.stop()
 
     def test_wait_as_b_send_completion(self):
-        self.assertRaises(Snap7Exception, self.partner.wait_as_b_send_completion)
+        self.assertRaises(RuntimeError, self.partner.wait_as_b_send_completion)
 
 
 class TestLibraryIntegration(unittest.TestCase):