Added tests to motoroutput (100%)

tests/test_motoroutput.py

@@ -0,0 +1,48 @@
+from unittest import TestCase, mock
+
+from visualswarm.control import motoroutput
+
+
+class MotorInterfaceTest(TestCase):
+
+    @mock.patch('visualswarm.env.EXIT_CONDITION', True)
+    @mock.patch('dbus.mainloop.glib.DBusGMainLoop', return_value=None)
+    @mock.patch('dbus.SessionBus')
+    @mock.patch('dbus.Interface')
+    def test_control_thymio(self, mock_network_init, mock_dbus_sessionbus, mock_dbus_init):
+        mock_network = mock.MagicMock(return_value=None)
+        mock_network.SetVariable.return_value = None
+        mock_network_init.return_value = mock_network
+
+        mock_dbus_sessionbus.return_value.get_object.return_value = None
+        control_stream = mock.MagicMock()
+        control_stream.get.return_value = (1, 1)
+
+        # Case 1 : healthy connection via asebamedulla
+        with mock.patch('visualswarm.control.motorinterface.asebamedulla_health') as mock_health:
+            mock_health.return_value = True
+            # Case 1/a : with no control
+            motoroutput.control_thymio(control_stream, with_control=False)
+            control_stream.get.assert_called_once()
+
+            control_stream.get.reset_mock()
+            mock_health.reset_mock()
+            # Case 1/b : with control
+            motoroutput.control_thymio(control_stream, with_control=True)
+            mock_dbus_init.assert_called_once_with(set_as_default=True)
+            mock_dbus_sessionbus.assert_called_once()
+            mock_network_init.assert_called_once()
+            mock_health.assert_called_once()
+            control_stream.get.assert_called_once()
+            self.assertEqual(mock_network.SetVariable.call_count, 2)
+
+        # Case 2 : unhealthy connection via asebamedulla
+        with mock.patch('visualswarm.control.motorinterface.asebamedulla_health') as mock_health:
+            with mock.patch('visualswarm.control.motorinterface.asebamedulla_end'):
+                mock_health.return_value = False
+                control_stream.reset_mock()
+                motoroutput.control_thymio(control_stream, with_control=False)
+                control_stream.get.assert_called_once()
+
+                with self.assertRaises(Exception):
+                    motoroutput.control_thymio(control_stream, with_control=True)

visualswarm/control/motoroutput.py

@@ -4,6 +4,7 @@
 
 from visualswarm.control import motorinterface
 from visualswarm.contrib import logparams, control
+from visualswarm import env
 
 # import tempfile
 # import random
@@ -50,6 +51,10 @@ def control_thymio(control_stream, with_control=False):
         # simply consuming the input stream so that we don't fill up memory
         while True:
             (v, dpsi) = control_stream.get()
+
+            # To test infinite loops
+            if env.EXIT_CONDITION:
+                break
     else:
         # Initializing DBus
         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
@@ -75,6 +80,10 @@ def control_thymio(control_stream, with_control=False):
                 network.SetVariable("thymio-II", "motor.right.target", [v_right])
 
                 logger.info(f"left: {v_left} \t right: {v_right}")
+
+                # To test infinite loops
+                if env.EXIT_CONDITION:
+                    break
         else:
             logger.error(f'{bcolors.FAIL}🗴 CONNECTION FAILED{bcolors.ENDC} via asebamedulla')
             motorinterface.asebamedulla_end()