Remove asserts from non tests (#250)

* Replace assert statements with if/else statements
Fix for #215

* Change ValueErrors to better type TypeErrors

* Fix wrong isinstance request back to "not in" check
Fix tests to await correct Error

* Fix wrong Error types
Replace "is False" with a better "not" statement

* Fix wrong errortype

* Use Snap7Exception to use unified error messages

* Fix expected exception to Snap7Exception to be consistent.

* Replace TypeError with Snap7Exception to seperate python- and snap7 type errors

* Replace Snap7Exceptions with better ValueError

* Change Snap7Exception with excepted ValueError

* Remove unuesed imports
replace Snap7Exception with ValueError

Author: swamper123

snap7/client.py

@@ -244,8 +244,9 @@ def read_area(self, area: str, dbnumber: int, start: int, size: int) -> bytearra
         :param start: offset to start writing
         :param size: number of units to read
         """
-        assert area in snap7.types.areas.values()
-        if area == snap7.types.S7AreaTM:
+        if area not in snap7.types.areas.values():
+            raise ValueError(f"{area} is not implemented in snap7.types")
+        elif area == snap7.types.S7AreaTM:
             wordlen = snap7.types.S7WLTimer
         elif area == snap7.types.S7AreaCT:
             wordlen = snap7.types.S7WLCounter
@@ -349,7 +350,8 @@ def get_block_info(self, blocktype: str, db_number: int) -> TS7BlockInfo:
     @error_wrap
     def set_session_password(self, password: str) -> int:
         """Send the password to the PLC to meet its security level."""
-        assert len(password) <= 8, 'maximum password length is 8'
+        if len(password) > 8:
+            raise ValueError("Maximum password length is 8")
         return self._library.Cli_SetSessionPassword(self._pointer,
                                                     c_char_p(password.encode()))
 
@@ -367,7 +369,8 @@ def set_connection_params(self, address: str, local_tsap: int, remote_tsap: int)
         :param local_tsap: Local TSAP (PC TSAP)
         :param remote_tsap: Remote TSAP (PLC TSAP)
         """
-        assert re.match(ipv4, address), f'{address} is invalid ipv4'
+        if not re.match(ipv4, address):
+            raise ValueError(f"{address} is invalid ipv4")
         result = self._library.Cli_SetConnectionParams(self._pointer, address,
                                                        c_uint16(local_tsap),
                                                        c_uint16(remote_tsap))
@@ -620,8 +623,8 @@ def wait_as_completion(self, timeout: int) -> int:
 
     def _prepare_as_read_area(self, area: str, size: int) -> Tuple[int, Array]:
         if area not in snap7.types.areas.values():
-            raise NotImplementedError(f"{area} is not implemented in snap7.types")
-        if area == snap7.types.S7AreaTM:
+            raise ValueError(f"{area} is not implemented in snap7.types")
+        elif area == snap7.types.S7AreaTM:
             wordlen = snap7.types.S7WLTimer
         elif area == snap7.types.S7AreaCT:
             wordlen = snap7.types.S7WLCounter
@@ -649,8 +652,8 @@ def as_read_area(self, area: str, dbnumber: int, start: int, size: int, wordlen:
 
     def _prepare_as_write_area(self, area: str, data: bytearray) -> Tuple[int, Array]:
         if area not in snap7.types.areas.values():
-            raise NotImplementedError(f"{area} is not implemented in snap7.types")
-        if area == snap7.types.S7AreaTM:
+            raise ValueError(f"{area} is not implemented in snap7.types")
+        elif area == snap7.types.S7AreaTM:
             wordlen = snap7.types.S7WLTimer
         elif area == snap7.types.S7AreaCT:
             wordlen = snap7.types.S7WLCounter

snap7/common.py

@@ -73,7 +73,8 @@ def error_text(error, context: str = "client") -> bytes:
     :param context: server, client or partner
     :returns: the error string
     """
-    assert context in ("client", "server", "partner")
+    if context not in ("client", "server", "partner"):
+        raise TypeError(f"Unkown context {context} used, should be either client, server or partner")
     logger.debug(f"error text for {hex(error)}")
     len_ = 1024
     text_type = c_char * len_

snap7/logo.py

@@ -248,7 +248,8 @@ def set_connection_params(self, ip_address: str, tsap_snap7: int, tsap_logo: int
         :param tsap_snap7: TSAP SNAP7 Client (e.g. 10.00 = 0x1000)
         :param tsap_logo: TSAP Logo Server (e.g. 20.00 = 0x2000)
         """
-        assert re.match(ipv4, ip_address), f'{ip_address} is invalid ipv4'
+        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))

snap7/partner.py

@@ -209,8 +209,11 @@ def start_to(self, local_ip: str, remote_ip: str, local_tsap: int, remote_tsap:
         :param local_tsap: Local TSAP
         :param remote_tsap: PLC TSAP
         """
-        assert re.match(ipv4, local_ip), f'{local_ip} is invalid ipv4'
-        assert re.match(ipv4, remote_ip), f'{remote_ip} is invalid ipv4'
+
+        if not re.match(ipv4, local_ip):
+            raise ValueError(f"{local_ip} is invalid ipv4")
+        if not re.match(ipv4, remote_ip):
+            raise ValueError(f"{remote_ip} is invalid ipv4")
         logger.info(f"starting partnering from {local_ip} to {remote_ip}")
         return self._library.Par_StartTo(self._pointer, local_ip, remote_ip,
                                          c_uint16(local_tsap),

snap7/server.py

@@ -219,7 +219,8 @@ def start_to(self, ip: str, tcpport: int = 102):
         if tcpport != 102:
             logger.info(f"setting server TCP port to {tcpport}")
             self.set_param(snap7.types.LocalPort, tcpport)
-        assert re.match(ipv4, ip), f'{ip} is invalid ipv4'
+        if not re.match(ipv4, ip):
+            raise ValueError(f"{ip} is invalid ipv4")
         logger.info(f"starting server to {ip}:102")
         return self.library.Srv_StartTo(self.pointer, ip)
 
@@ -242,7 +243,8 @@ def set_mask(self, kind: int, mask: int):
     def set_cpu_status(self, status: int):
         """Sets the Virtual CPU status.
         """
-        assert status in snap7.types.cpu_statuses, f'unknown cpu state {status}'
+        if status not in snap7.types.cpu_statuses:
+            raise ValueError(f"The cpu state ({status}) is invalid")
         logger.debug(f"setting cpu status to {status}")
         return self.library.Srv_SetCpuStatus(self.pointer, status)
 

snap7/util.py

@@ -106,7 +106,9 @@ def set_bool(bytearray_: bytearray, byte_index: int, bool_index: int, value: boo
     """
     Set boolean value on location in bytearray
     """
-    assert value in [0, 1, True, False]
+    if value not in {0, 1, True, False}:
+        raise TypeError(f"Value value:{value} is not a boolean expression.")
+
     current_value = get_bool(bytearray_, byte_index, bool_index)
     index_value = 1 << bool_index
 
@@ -200,7 +202,8 @@ def set_string(bytearray_: bytearray, byte_index: int, value: str, max_size: int
     :params value: string data
     :params max_size: max possible string size
     """
-    assert isinstance(value, str)
+    if not isinstance(value, str):
+        raise TypeError(f"Value value:{value} is not from Type string")
 
     size = len(value)
     # FAIL HARD WHEN trying to write too much data into PLC
@@ -487,7 +490,8 @@ def __len__(self):
         return len(self.index)
 
     def set_data(self, bytearray_: bytearray):
-        assert (isinstance(bytearray_, bytearray))
+        if not isinstance(bytearray_, bytearray):
+            raise TypeError(f"Value bytearray_: {bytearray_} is not from type bytearray")
         self._bytearray = bytearray_
 
 
@@ -505,7 +509,8 @@ def __init__(self, bytearray_, _specification, row_size=0, db_offset=0, layout_o
         self.row_size = row_size
         self.row_offset = row_offset  # start of writable part of row
 
-        assert (isinstance(bytearray_, (bytearray, DB)))
+        if not isinstance(bytearray_, (bytearray, DB)):
+            raise TypeError(f"Value bytearray_ {bytearray_} is not from type (bytearray, DB)")
         self._bytearray = bytearray_
         self._specification = parse_specification(_specification)
 
@@ -530,12 +535,10 @@ def __getitem__(self, key):
         """
         Get a specific db field
         """
-        assert key in self._specification
         index, _type = self._specification[key]
         return self.get_value(index, _type)
 
     def __setitem__(self, key, value):
-        assert key in self._specification
         index, _type = self._specification[key]
         self.set_value(index, _type, value)
 
@@ -675,8 +678,10 @@ def write(self, client):
         """
         Write current data to db in plc
         """
-        assert (isinstance(self._bytearray, DB))
-        assert (self.row_size >= 0)
+        if not isinstance(self._bytearray, DB):
+            raise TypeError(f"Value self._bytearray: {self._bytearray} is not from type DB.")
+        if self.row_size < 0:
+            raise ValueError("row_size must be greater equal zero.")
 
         db_nr = self._bytearray.db_number
         offset = self.db_offset
@@ -694,8 +699,10 @@ def read(self, client):
         """
         read current data of db row from plc
         """
-        assert (isinstance(self._bytearray, DB))
-        assert (self.row_size >= 0)
+        if not isinstance(self._bytearray, DB):
+            raise TypeError(f"Value self._bytearray:{self._bytearray} is not from type DB.")
+        if self.row_size < 0:
+            raise ValueError("row_size must be greater equal zero.")
         db_nr = self._bytearray.db_number
         bytearray_ = client.db_read(db_nr, self.db_offset, self.row_size)
 

test/test_server.py

@@ -67,7 +67,7 @@ def test_set_cpu_status(self):
         self.server.set_cpu_status(0)
         self.server.set_cpu_status(4)
         self.server.set_cpu_status(8)
-        self.assertRaises(AssertionError, self.server.set_cpu_status, -1)
+        self.assertRaises(ValueError, self.server.set_cpu_status, -1)
 
     def test_set_mask(self):
         self.server.set_mask(kind=snap7.types.mkEvent, mask=10)
@@ -115,7 +115,7 @@ def test_clear_events(self):
 
     def test_start_to(self):
         self.server.start_to('0.0.0.0')
-        self.assertRaises(AssertionError, self.server.start_to, 'bogus')
+        self.assertRaises(ValueError, self.server.start_to, 'bogus')
 
     def test_get_param(self):
         # check the defaults