Fixed tests and increased coverage

Author: giuseppe-carboni

.coveragerc

@@ -7,7 +7,7 @@ exclude_lines =
     if __name__ == '__main__':
     self.fail\(.*\)
     except KeyboardInterrupt:
-    except ImportError:
+    except ImportError.*
     skip coverage
 include =
     ./*

scripts/discos-simulator

@@ -22,35 +22,36 @@ AVAILABLE_SIMULATORS = list_simulators()
 
 def system_from_arg(system_name):
     try:
-        return importlib.import_module('simulators.%s' % system_name)
+        return importlib.import_module(f"simulators.{system_name}")
     except ImportError as e:
         error = e.args[0]
-        if 'No module named' in error:
-            error = 'System "%s" unavailable.' % system_name
+        if "No module named" in error:
+            error = f"System '{system_name}' unavailable."
         else:
             error = error % system_name
-        raise ArgumentTypeError(error)
+        raise ArgumentTypeError(error) from e
+
 
 parser = ArgumentParser()
 parser.add_argument(
-    'action',
-    choices=['start', 'stop', 'list']
+    "action",
+    choices=["start", "stop", "list"]
 )
 parser.add_argument(
-    '-s', '--system',
+    "-s", "--system",
     type=system_from_arg,
     required=False,
-    help='System name: active_surface, acu, ...',
+    help="System name: active_surface, acu, ...",
 )
 parser.add_argument(
-    '-t', '--type',
+    "-t", "--type",
     type=str,
     required=False,
-    help='System configuration type: IFD_14_channels for if_distributor, ...',
+    help="System configuration type: IFD_14_channels for if_distributor, ...",
 )
 args = parser.parse_args()
 
-if args.action == 'list':
+if args.action == "list":
     print(
         "Available simulators: '"
         + "', '".join(AVAILABLE_SIMULATORS)
@@ -60,55 +61,50 @@ if args.action == 'list':
 kwargs = {}
 
 if args.type:
+    name = args.system.__name__.rsplit(".", 1)[1]
     if not args.system:
         parser.error(
             "The '--type' argument only has to be used "
             + "in conjunction with the '--system' argument."
         )
     try:
-        systems = getattr(args.system, 'systems')
+        systems = getattr(args.system, "systems")
     except AttributeError:
         parser.error(
-            ('System %s has no configurations other than the default one. '
-            % args.system.__name__.rsplit('.', 1)[1])
-            + "Omit the '--type' argument to start the simulator properly."
+            f"System '{name}' has no configurations other than the default one"
+            + ". Omit the '--type' argument to start the simulator properly."
         )
     if args.type not in systems:
         err_string = (
-            'Configuration %s for system %s not found.\n' %
-            (args.type, args.system.__name__.rsplit('.', 1)[1])
+            f"Configuration '{args.type}' for system '{name}' not found.\n"
         )
-        err_string += 'Available configurations for chosen simulator:\n'
+        err_string += "Available configurations for desired simulator:\n"
         err_string += "'" + "', ".join(systems) + "'."
         parser.error(err_string)
-    kwargs['system_type'] = args.type
+    kwargs["system_type"] = args.type
 
-if args.action == 'start':
+if args.action == "start":
     if args.system:
         try:
             simulator = Simulator(args.system, **kwargs)
             simulator.start()
         except socket.error:
+            name = args.system.__name__.rsplit(".", 1)[1]
             print(
-                "Cannot start simulator '%s', address already in use." %
-                args.system.__name__.rsplit('.', 1)[1]
+                f"Cannot start simulator '{name}', address already in use."
             )
             sys.exit(1)
     else:
         for sim in AVAILABLE_SIMULATORS:
-            subprocess.Popen([
-                sys.argv[0],
-                'start',
-                '-s',
-                sim
-            ])
+            with subprocess.Popen([sys.argv[0], "start", "-s", sim]) as _:
+                pass
         time.sleep(3)
-elif args.action == 'stop':
+elif args.action == "stop":
     if args.system:
         simulator = Simulator(args.system, **kwargs)
         simulator.stop()
     else:
         for sim in AVAILABLE_SIMULATORS:
-            sim = importlib.import_module('simulators.%s' % sim)
+            sim = importlib.import_module(f"simulators.{sim}")
             simulator = Simulator(sim, **kwargs)
             simulator.stop()

simulators/acu/pointing_status.py

@@ -6,12 +6,12 @@
 from datetime import datetime, timedelta
 try:
     import numpy as np
-except ImportError as ex:  # skip coverage
+except ImportError as ex:
     raise ImportError('The `numpy` package, required for the simulator'
         + ' to run, is missing!') from ex
 try:
     from scipy import interpolate
-except ImportError as ex:  # skip coverage
+except ImportError as ex:
     raise ImportError('The `scipy` package, required for the simulator'
         + ' to run, is missing!') from ex
 from simulators import utils

simulators/minor_servos/__init__.py

@@ -1,18 +1,18 @@
 import time
 import re
 import random
+import threading
 try:
     from numpy import sign
-except ImportError as ex:  # skip coverage
+except ImportError as ex:
     raise ImportError('The `numpy` package, required for the simulator'
         + ' to run, is missing!') from ex
 try:
     from scipy.interpolate import splrep, splev
-except ImportError as ex:  # skip coverage
+except ImportError as ex:
     raise ImportError('The `scipy` package, required for the simulator'
         + ' to run, is missing!') from ex
 from ctypes import c_bool, c_int
-from threading import Lock, Thread, Timer
 from multiprocessing import Value
 from bisect import bisect_left
 from socketserver import ThreadingTCPServer
@@ -104,7 +104,7 @@ def __init__(self, timer_value=DEFAULT_TIMER_VALUE, rest_api=True):
             self.configurations
         )
         self.stop = Value(c_bool, False)
-        self.update_thread = Thread(
+        self.update_thread = threading.Thread(
             target=self._update,
             args=(self.stop, self.servos)
         )
@@ -116,11 +116,11 @@ def __init__(self, timer_value=DEFAULT_TIMER_VALUE, rest_api=True):
                 httpserver_address,
                 VBrainRequestHandler
             )
-            self.server_thread = Thread(
+            self.server_thread = threading.Thread(
                 target=self.httpserver.serve_forever
             )
             self.server_thread.start()
-        self.cover_timer = Timer(1, lambda: None)
+        self.cover_timer = threading.Timer(1, lambda: None)
 
     def __del__(self):
         self.system_stop()
@@ -218,7 +218,7 @@ def _setup(self, args):
                 and self.gregorian_cap.value != gregorian_cap_position):
             self.cover_timer.cancel()
             _change_atomic_value(self.gregorian_cap, 0)
-            self.cover_timer = Timer(
+            self.cover_timer = threading.Timer(
                 self.timer_value,
                 _change_atomic_value,
                 args=(self.gregorian_cap, gregorian_cap_position)
@@ -245,7 +245,7 @@ def _stow(self, args):
                 self.cover_timer.cancel()
                 if self.gregorian_cap.value <= 1 or stow_pos == 1:
                     _change_atomic_value(self.gregorian_cap, 0)
-                    self.cover_timer = Timer(
+                    self.cover_timer = threading.Timer(
                         self.timer_value,
                         _change_atomic_value,
                         args=(self.gregorian_cap, stow_pos)
@@ -261,7 +261,7 @@ def _stow(self, args):
             servo = self.servos.get(servo_id)
             servo.operative_mode_timer.cancel()
             _change_atomic_value(servo.operative_mode, 0)
-            servo.operative_mode_timer = Timer(
+            servo.operative_mode_timer = threading.Timer(
                 self.timer_value,
                 _change_atomic_value,
                 args=(servo.operative_mode, 20)  # STOW
@@ -451,9 +451,9 @@ def __init__(self, name, dof=1):
         self.cmd_coords = self.coords.copy()
         self.offsets = [0] * self.DOF
         self.last_status_read = 0
-        self.operative_mode_timer = Timer(1, lambda: None)
+        self.operative_mode_timer = threading.Timer(1, lambda: None)
         if self.program_track_capable:
-            self.trajectory_lock = Lock()
+            self.trajectory_lock = threading.Lock()
             self.trajectory_id = None
             self.trajectory_start_time = None
             self.trajectory_point_id = None

tests/test_minor_servos.py

@@ -3,10 +3,10 @@
 import time
 import requests
 from simulators.minor_servos import System, httpserver_address
+from simulators.minor_servos import DEFAULT_TIMER_VALUE
 from simulators.minor_servos.helpers import VBrainRequestHandler
 from simulators.utils import FastTimeMock
 
-TIMER_VALUE = 0.01
 tail = '\r\n'
 good = r'^OUTPUT:GOOD,[0-9]+\.[0-9]{6}'
 bad = f'^OUTPUT:BAD{tail}$'
@@ -16,7 +16,6 @@ class TestMinorServos(unittest.TestCase):
 
     def setUp(self):
         self.system = System(
-            timer_value=TIMER_VALUE,
             rest_api=False
         )
 
@@ -220,35 +219,44 @@ def test_setup_unknown_configuration(self):
         self.assertRegex(self.system.parse(cmd[-1]), bad)
 
     def test_stow(self):
+        fast_time = FastTimeMock(100)
+        fast_time.start()
         for servo_id, servo in self.system.servos.items():
             cmd = f'STOW={servo_id},1{tail}'
             for byte in cmd[:-1]:
                 self.assertTrue(self.system.parse(byte))
             self.assertRegex(self.system.parse(cmd[-1]), f'{good}{tail}$')
-            time.sleep(TIMER_VALUE + 0.1)
+            time.sleep(DEFAULT_TIMER_VALUE + 1)
             self.assertEqual(servo.operative_mode.value, 20)  # STOW mode
+        fast_time.stop()
 
     def test_stow_gregorian_cap(self):
+        fast_time = FastTimeMock(100)
+        fast_time.start()
         for stow_pos in [1, 2]:
             cmd = f'STOW=GREGORIAN_CAP,{stow_pos}{tail}'
             for byte in cmd[:-1]:
                 self.assertTrue(self.system.parse(byte))
             self.assertRegex(self.system.parse(cmd[-1]), f'{good}{tail}$')
-            time.sleep(TIMER_VALUE + 0.1)
+            time.sleep(DEFAULT_TIMER_VALUE + 1)
             self.assertEqual(self.system.gregorian_cap.value, stow_pos)
+        fast_time.stop()
 
     def test_stow_gregorian_air_blade(self):
+        fast_time = FastTimeMock(100)
+        fast_time.start()
         cmd = f'STOW=GREGORIAN_CAP,2{tail}'
         for byte in cmd[:-1]:
             self.assertTrue(self.system.parse(byte))
         self.assertRegex(self.system.parse(cmd[-1]), f'{good}{tail}$')
-        time.sleep(TIMER_VALUE + 0.1)
+        time.sleep(DEFAULT_TIMER_VALUE + 1)
         self.assertEqual(self.system.gregorian_cap.value, 2)
         cmd = f'STOW=GREGORIAN_CAP,3{tail}'
         for byte in cmd[:-1]:
             self.assertTrue(self.system.parse(byte))
         self.assertRegex(self.system.parse(cmd[-1]), f'{good}{tail}$')
         self.assertEqual(self.system.gregorian_cap.value, 3)
+        fast_time.stop()
 
     def test_stow_gregorian_cap_wrong_pos(self):
         cmd = f'STOW=GREGORIAN_CAP,5{tail}'
@@ -288,7 +296,7 @@ def test_stop(self):
             for byte in cmd[:-1]:
                 self.assertTrue(self.system.parse(byte))
             self.assertRegex(self.system.parse(cmd[-1]), f'{good}{tail}$')
-            time.sleep(TIMER_VALUE + 0.1)
+            time.sleep(0.2)
             self.assertEqual(servo.operative_mode.value, 30)  # STOP mode
 
     def test_stop_no_arguments(self):
@@ -601,7 +609,7 @@ def test_rest_GET(self):
             baseurl = f'http://{httpserver_address[0]}:{httpserver_address[1]}'
             url = baseurl + url
             try:
-                response = requests.get(url, timeout=TIMER_VALUE)
+                response = requests.get(url, timeout=0.2)
                 self.assertEqual(response.status_code, 200)
                 self.assertEqual(response.json(), VBrainRequestHandler.answer)
             except requests.exceptions.ReadTimeout:
@@ -611,7 +619,7 @@ def test_rest_GET_wrong_address(self):
         baseurl = f'http://{httpserver_address[0]}:{httpserver_address[1]}'
         url = baseurl + '/wrong'
         try:
-            response = requests.get(url, timeout=TIMER_VALUE)
+            response = requests.get(url, timeout=0.2)
             self.assertEqual(response.status_code, 404)
         except requests.exceptions.ReadTimeout:
             self.fail('Request is taking too long to be answered')