test: improve code coverage

Author: feeph

feeph/i2c/burst_handler.py

@@ -66,15 +66,15 @@ def read_register(self, register: int, byte_count: int = 1, max_tries: int = 5)
         buf_r = bytearray(byte_count)
         buf_r[0] = register
         buf_w = bytearray(byte_count)
+        cur_try = 0
         for cur_try in range(1, 1 + max_tries):
             try:
                 self._i2c_bus.writeto_then_readfrom(address=self._i2c_adr, buffer_out=buf_r, buffer_in=buf_w)
                 return convert_bytearry_to_uint(buf_w)
-            except OSError as e:
+            # protect against sporadic errors on actual devices
+            # (maybe we can do something to prevent these errors?)
+            except (OSError, RuntimeError) as e:
                 # [Errno 121] Remote I/O error
-                LH.warning("[%s] Failed to read register 0x%02X (%i/%i): %s",  __name__, register, cur_try, max_tries, e)
-                time.sleep(0.001)
-            except RuntimeError as e:
                 LH.warning("[%s] Unable to read register 0x%02X (%i/%i): %s", __name__, register, cur_try, max_tries, e)
                 time.sleep(0.001)
         raise RuntimeError(f"Unable to read register 0x{register:02X} after {cur_try} attempts. Giving up.")
@@ -93,16 +93,16 @@ def write_register(self, register: int, value: int, byte_count: int = 1, max_tri
         buf = bytearray([register])
         for byte in ba:
             buf.append(byte)
+        cur_try = 0
         for cur_try in range(1, 1 + max_tries):
             try:
                 self._i2c_bus.writeto(address=self._i2c_adr, buffer=buf)
                 return
-            except OSError as e:
+            # protect against sporadic errors on actual devices
+            # (maybe we can do something to prevent these errors?)
+            except (OSError, RuntimeError) as e:
                 # [Errno 121] Remote I/O error
-                LH.warning("[%s] Failed to read register 0x%02X (%i/%i): %s",  __name__, register, cur_try, max_tries, e)
-                time.sleep(0.1)
-            except RuntimeError as e:
-                LH.warning("[%s] Unable to read register 0x%02X (%i/%i): %s", __name__, register, cur_try, max_tries, e)
+                LH.warning("[%s] Unable to write register 0x%02X (%i/%i): %s", __name__, register, cur_try, max_tries, e)
                 time.sleep(0.1)
         raise RuntimeError(f"Unable to read register 0x{register:02X} after {cur_try} attempts. Giving up.")
 
@@ -121,15 +121,15 @@ def get_state(self, byte_count: int = 1, max_tries: int = 5) -> int:
             LH.warning("Multi byte reads are not implemented yet! Returning a single byte instead.")
             byte_count = 1
         buf = bytearray(byte_count)
+        cur_try = 0
         for cur_try in range(1, 1 + max_tries):
             try:
                 self._i2c_bus.readfrom_into(address=self._i2c_adr, buffer=buf)
                 return buf[0]
-            except OSError as e:
+            # protect against sporadic errors on actual devices
+            # (maybe we can do something to prevent these errors?)
+            except (OSError, RuntimeError) as e:
                 # [Errno 121] Remote I/O error
-                LH.warning("[%s] Failed to read state (%i/%i): %s",  __name__, cur_try, max_tries, e)
-                time.sleep(0.001)
-            except RuntimeError as e:
                 LH.warning("[%s] Unable to read state (%i/%i): %s", __name__, cur_try, max_tries, e)
                 time.sleep(0.001)
         raise RuntimeError(f"Unable to read state after {cur_try} attempts. Giving up.")
@@ -145,15 +145,15 @@ def set_state(self, value: int, byte_count: int = 1, max_tries: int = 3):
             LH.warning("Multi byte writes are not implemented yet! Returning a single byte instead.")
             byte_count = 1
         buf = convert_uint_to_bytearry(value, byte_count)
+        cur_try = 0
         for cur_try in range(1, 1 + max_tries):
             try:
                 self._i2c_bus.writeto(address=self._i2c_adr, buffer=buf)
                 return
-            except OSError as e:
+            # protect against sporadic errors on actual devices
+            # (maybe we can do something to prevent these errors?)
+            except (OSError, RuntimeError) as e:
                 # [Errno 121] Remote I/O error
-                LH.warning("[%s] Failed to write state (%i/%i): %s",  __name__, cur_try, max_tries, e)
-                time.sleep(0.1)
-            except RuntimeError as e:
                 LH.warning("[%s] Unable to write state (%i/%i): %s", __name__, cur_try, max_tries, e)
                 time.sleep(0.1)
         raise RuntimeError(f"Unable to write state after {cur_try} attempts. Giving up.")

feeph/i2c/emulation.py

@@ -55,12 +55,18 @@ def unlock(self):
 
     # replicate the signature of busio.I2C
     # pylint: disable=too-many-arguments
-    def readfrom_into(self, address, buffer, *, start=0, end=None, stop=True):
+    def readfrom_into(self, address: int, buffer: bytearray, *, start=0, end=None, stop=True):
         """
         read device state
 
         (buffer is used as an output parameter)
         """
+        # make sure that buffer truly is a bytearray
+        # (we really want this to be a bytearray because bytearray performs
+        # its own input validation and automatically guarantees we're not
+        # getting negative byte values or other unexpected data as input)
+        if not isinstance(buffer, bytearray):
+            raise ValueError("buffer must be of type 'bytearray'")
         i2c_device_address  = address
         i2c_device_register = -1
         value = self._state[i2c_device_address][i2c_device_register]
@@ -76,6 +82,12 @@ def writeto(self, address: int, buffer: bytearray, *, start=0, end=None):
         """
         write device state or register
         """
+        # make sure that buffer truly is a bytearray
+        # (we really want this to be a bytearray because bytearray performs
+        # its own input validation and automatically guarantees we're not
+        # getting negative byte values or other unexpected data as input)
+        if not isinstance(buffer, bytearray):
+            raise ValueError("buffer must be of type 'bytearray'")
         if len(buffer) == 1:
             # device status
             i2c_device_address  = address
@@ -85,8 +97,6 @@ def writeto(self, address: int, buffer: bytearray, *, start=0, end=None):
             # device register
             i2c_device_address  = address
             i2c_device_register = buffer[0]
-            if i2c_device_register < 0:
-                raise ValueError("device register can't be negative")
             value = convert_bytearry_to_uint(buffer[1:])
         self._state[i2c_device_address][i2c_device_register] = value
 
@@ -96,10 +106,16 @@ def writeto_then_readfrom(self, address: int, buffer_out: bytearray, buffer_in:
 
         (buffer_in is used as an output parameter)
         """
+        # make sure that buffer_in and buffer_out truly are a bytearray
+        # (we really want this to be a bytearray because bytearray performs
+        # its own input validation and automatically guarantees we're not
+        # getting negative byte values or other unexpected data as input)
+        if not isinstance(buffer_in, bytearray):
+            raise ValueError("buffer_in must be of type 'bytearray'")
+        if not isinstance(buffer_out, bytearray):
+            raise ValueError("buffer_out must be of type 'bytearray'")
         i2c_device_address  = address
         i2c_device_register = buffer_out[0]
-        if i2c_device_register < 0:
-            raise ValueError("device register can't be negative")
         value = self._state[i2c_device_address][i2c_device_register]
         ba = convert_uint_to_bytearry(value, len(buffer_in))
         # copy computed result to output parameter

feeph/i2c/utility.py

@@ -8,7 +8,7 @@
 import feeph.i2c as sut  # sytem under test
 
 
-# pylint: disable=protected-access
+# pylint: disable=protected-access,too-many-public-methods
 class TestBurstHandler(unittest.TestCase):
 
     def test_read_device_register(self):
@@ -39,6 +39,15 @@ def test_read_device_register_multibyte(self):
         # -----------------------------------------------------------------
         self.assertEqual(computed, expected)
 
+    def test_read_device_register_insufficient_tries(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
+            # under realistic circumstances the max_tries would be a positive
+            # value but we're intentionally setting it to 0 to force an error
+            self.assertRaises(RuntimeError, bh.read_register, 0x00, max_tries=0)
+
     def test_read_device_registers(self):
         state = {
             0x4C: {
@@ -105,6 +114,15 @@ def test_write_device_register_multibyte(self):
         # -----------------------------------------------------------------
         self.assertEqual(computed, expected)
 
+    def test_write_device_register_insufficient_tries(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
+            # under realistic circumstances the max_tries would be a positive
+            # value but we're intentionally setting it to 0 to force an error
+            self.assertRaises(RuntimeError, bh.write_register, 0x00, value=0x12, max_tries=0)
+
     def test_write_device_registers(self):
         state = {
             0x4C: {
@@ -188,6 +206,15 @@ def test_get_state_multibyte(self):
         # -----------------------------------------------------------------
         self.assertEqual(computed, expected)
 
+    def test_get_state_insufficient_tries(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
+            # under realistic circumstances the max_tries would be a positive
+            # value but we're intentionally setting it to 0 to force an error
+            self.assertRaises(RuntimeError, bh.get_state, max_tries=0)
+
     def test_set_state(self):
         state = {
             0x70: {-1: 0x00},
@@ -206,11 +233,49 @@ def test_set_state_multibyte(self):
         # -----------------------------------------------------------------
         # -----------------------------------------------------------------
         with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x70) as bh:
-            self.assertRaises(ValueError, bh.set_state, value=0x0102)
+            self.assertRaises(ValueError, bh.set_state, value=0x0102, byte_count=2)
+
+    def test_set_state_insufficient_tries(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
+            # under realistic circumstances the max_tries would be a positive
+            # value but we're intentionally setting it to 0 to force an error
+            self.assertRaises(RuntimeError, bh.set_state, value=0x12, max_tries=0)
 
     # ---------------------------------------------------------------------
 
-    def test_no_timeout(self):
+    def test_invalid_device_address(self):
+        # this code tests the equivalent of:
+        # with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0xFFFF) as bh:
+        #     ...
+        i2c_bus = sut.EmulatedI2C(state={})
+        bh = sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0xFFFF)
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, bh.__enter__)
+
+    def test_invalid_device_register(self):
+        # this code tests the equivalent of:
+        # with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0xFFFF) as bh:
+        #     ...
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x70) as bh:
+            self.assertRaises(ValueError, bh.read_register, register=0xFFFF)
+
+    def test_invalid_timeout(self):
+        # this code tests the equivalent of:
+        # with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C, timeout_ms=0) as bh:
+        #     ...
+        i2c_bus = sut.EmulatedI2C(state={}, lock_chance=1)
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, sut.BurstHandler, i2c_bus=i2c_bus, i2c_adr=0x4C, timeout_ms=0)
+
+    def test_hard_to_lock(self):
         state = {
             0x4C: {
                 0x00: 0x12,
@@ -225,3 +290,13 @@ def test_no_timeout(self):
         expected = 0x12
         # -----------------------------------------------------------------
         self.assertEqual(computed, expected)
+
+    def test_unable_to_lock(self):
+        # this code tests the equivalent of:
+        # with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
+        #     ...
+        i2c_bus = sut.EmulatedI2C(state={}, lock_chance=0)  # impossible to acquire a lock
+        bh = sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x4C)
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(RuntimeError, bh.__enter__)

tests/test_burst_handler.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+"""
+perform I²C bus related tests
+"""
+
+import unittest
+
+import feeph.i2c as sut  # sytem under test
+
+
+# pylint: disable=protected-access
+class TestEmulatedI2C(unittest.TestCase):
+
+    def test_input_validation1(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, i2c_bus.readfrom_into, 0x12, [-1, 0x00])
+
+    def test_input_validation2(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, i2c_bus.writeto, 0x12, [-1, 0x00])
+
+    def test_input_validation3(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, i2c_bus.writeto_then_readfrom, 0x12, bytearray(0), [-1, 0x00])
+
+    def test_input_validation4(self):
+        i2c_bus = sut.EmulatedI2C(state={})
+        # -----------------------------------------------------------------
+        # -----------------------------------------------------------------
+        self.assertRaises(ValueError, i2c_bus.writeto_then_readfrom, 0x12, [-1, 0x00], bytearray(0))