Feature/cli no UI

framework/python/src/core/test_runner.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 """Wrapper for the Testrun that simplifies
 virtual testing procedure by allowing direct calling
 from the command line.
@@ -25,6 +24,7 @@
 from testrun import Testrun
 from common import logger
 import signal
+import io
 
 LOGGER = logger.get_logger("runner")
 
@@ -37,13 +37,17 @@ def __init__(self,
                validate=False,
                net_only=False,
                single_intf=False,
-               no_ui=False):
+               no_ui=False,
+               target=None,
+               firmware=None):
     self._register_exits()
     self._testrun = Testrun(config_file=config_file,
                             validate=validate,
                             net_only=net_only,
                             single_intf=single_intf,
-                            no_ui=no_ui)
+                            no_ui=no_ui,
+                            target_mac=target,
+                            firmware=firmware)
 
   def _register_exits(self):
     signal.signal(signal.SIGINT, self._exit_handler)
@@ -73,8 +77,7 @@ def parse_args():
       "-f",
       "--config-file",
       default=None,
-      help="Define the configuration file for Testrun and Network Orchestrator"
-  )
+      help="Define the configuration file for Testrun and Network Orchestrator")
   parser.add_argument(
       "--validate",
       default=False,
@@ -91,7 +94,38 @@ def parse_args():
                       default=False,
                       action="store_true",
                       help="Do not launch the user interface")
+  parser.add_argument("--target",
+                      default=None,
+                      type=str,
+                      help="MAC address of the target device")
+  parser.add_argument("-fw",
+                      "--firmware",
+                      default=None,
+                      type=str,
+                      help="Firmware version to be tested")
+
   parsed_args = parser.parse_known_args()[0]
+
+  if (parsed_args.no_ui and not parsed_args.net_only
+      and (parsed_args.target is None or parsed_args.firmware is None)):
+    # Capture help text
+    help_text = io.StringIO()
+    parser.print_help(file=help_text)
+
+    # Get help text as lines and find where "Testrun" starts (skip usage)
+    help_lines = help_text.getvalue().splitlines()
+    start_index = next(
+        (i for i, line in enumerate(help_lines) if "Testrun" in line), 0)
+
+    # Join only lines starting from "Testrun" and print without extra newlines
+    help_message = "\n".join(line.rstrip() for line in help_lines[start_index:])
+    print(help_message)
+
+    print(
+        "Error: --target and --firmware are required when --no-ui is specified",
+        file=sys.stderr)
+    sys.exit(1)
+
   return parsed_args
 
 
@@ -101,4 +135,6 @@ def parse_args():
                       validate=args.validate,
                       net_only=args.net_only,
                       single_intf=args.single_intf,
-                      no_ui=args.no_ui)
+                      no_ui=args.no_ui,
+                      target=args.target,
+                      firmware=args.firmware)

framework/python/src/core/testrun.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 """The overall control of the Testrun application.
 This file provides the integration between all of the
 Testrun components, such as net_orc, test_orc and test_ui.
@@ -55,6 +54,7 @@
 
 MAX_DEVICE_REPORTS_KEY = 'max_device_reports'
 
+
 class Testrun:  # pylint: disable=too-few-public-methods
   """Testrun controller.
 
@@ -67,15 +67,17 @@ def __init__(self,
                validate=False,
                net_only=False,
                single_intf=False,
-               no_ui=False):
+               no_ui=False,
+               target_mac=None,
+               firmware=None):
 
     # Locate parent directory
     current_dir = os.path.dirname(os.path.realpath(__file__))
 
     # Locate the test-run root directory, 4 levels,
     # src->python->framework->test-run
-    self._root_dir = os.path.dirname(os.path.dirname(
-      os.path.dirname(os.path.dirname(current_dir))))
+    self._root_dir = os.path.dirname(
+        os.path.dirname(os.path.dirname(os.path.dirname(current_dir))))
 
     # Determine config file
     if config_file is None:
@@ -103,15 +105,26 @@ def __init__(self,
     if validate:
       self._session.add_runtime_param('validate')
 
-    self._net_orc = net_orc.NetworkOrchestrator(
-      session=self._session)
-    self._test_orc = test_orc.TestOrchestrator(
-      self._session,
-      self._net_orc)
+    self._net_orc = net_orc.NetworkOrchestrator(session=self._session)
+    self._test_orc = test_orc.TestOrchestrator(self._session, self._net_orc)
 
     # Load device repository
     self.load_all_devices()
 
+    # If no_ui selected and not network only mode,
+    # load the target device into the session
+    if self._no_ui and not net_only:
+      target_device = self._session.get_device(target_mac)
+      if target_device is not None:
+        target_device.firmware = firmware
+        self._session.set_target_device(target_device)
+      else:
+        print(
+            f'Target device specified does not exist in device registry: '
+            f'{target_mac}',
+            file=sys.stderr)
+        sys.exit(1)
+
     # Load test modules
     self._test_orc.start()
 
@@ -165,8 +178,7 @@ def _load_devices(self, device_dir):
 
     for device_folder in os.listdir(device_dir):
 
-      device_config_file_path = os.path.join(device_dir,
-                                             device_folder,
+      device_config_file_path = os.path.join(device_dir, device_folder,
                                              DEVICE_CONFIG)
 
       # Check if device config file exists before loading
@@ -183,7 +195,7 @@ def _load_devices(self, device_dir):
           device_config_json = json.load(device_config_file)
         except json.decoder.JSONDecodeError as e:
           LOGGER.error('Invalid JSON found in ' +
-            f'device configuration {device_config_file_path}')
+                       f'device configuration {device_config_file_path}')
           LOGGER.debug(e)
           continue
 
@@ -219,11 +231,11 @@ def _load_devices(self, device_dir):
 
         if DEVICE_ADDITIONAL_INFO_KEY in device_config_json:
           device.additional_info = device_config_json.get(
-            DEVICE_ADDITIONAL_INFO_KEY)
+              DEVICE_ADDITIONAL_INFO_KEY)
 
         if None in [device.type, device.technology, device.test_pack]:
           LOGGER.warning(
-            'Device is outdated and requires further configuration')
+              'Device is outdated and requires further configuration')
           device.status = 'Invalid'
 
         self._load_test_reports(device)
@@ -250,26 +262,20 @@ def _load_test_reports(self, device):
 
     for report_folder in os.listdir(reports_folder):
       # 1.3 file path
-      report_json_file_path = os.path.join(
-        reports_folder,
-        report_folder,
-        'test',
-        device.mac_addr.replace(':',''),
-        'report.json')
+      report_json_file_path = os.path.join(reports_folder, report_folder,
+                                           'test',
+                                           device.mac_addr.replace(':', ''),
+                                           'report.json')
 
       if not os.path.isfile(report_json_file_path):
         # Revert to pre 1.3 file path
-        report_json_file_path = os.path.join(
-          reports_folder,
-          report_folder,
-          'report.json')
+        report_json_file_path = os.path.join(reports_folder, report_folder,
+                                             'report.json')
 
       if not os.path.isfile(report_json_file_path):
         # Revert to pre 1.3 file path
-        report_json_file_path = os.path.join(
-          reports_folder,
-          report_folder,
-          'report.json')
+        report_json_file_path = os.path.join(reports_folder, report_folder,
+                                             'report.json')
 
       # Check if the report.json file exists
       if not os.path.isfile(report_json_file_path):
@@ -285,9 +291,8 @@ def _load_test_reports(self, device):
 
   def get_reports_folder(self, device):
     """Return the reports folder path for the device"""
-    return os.path.join(self._root_dir,
-                        LOCAL_DEVICES_DIR,
-                        device.device_folder, 'reports')
+    return os.path.join(self._root_dir, LOCAL_DEVICES_DIR, device.device_folder,
+                        'reports')
 
   def delete_report(self, device: Device, timestamp):
     LOGGER.debug(f'Deleting test report for device {device.model} ' +
@@ -308,15 +313,13 @@ def delete_report(self, device: Device, timestamp):
   def create_device(self, device: Device):
 
     # Define the device folder location
-    device_folder_path = os.path.join(self._root_dir,
-                                      LOCAL_DEVICES_DIR,
+    device_folder_path = os.path.join(self._root_dir, LOCAL_DEVICES_DIR,
                                       device.device_folder)
 
     # Create the directory
     os.makedirs(device_folder_path)
 
-    config_file_path = os.path.join(device_folder_path,
-                                    DEVICE_CONFIG)
+    config_file_path = os.path.join(device_folder_path, DEVICE_CONFIG)
 
     with open(config_file_path, 'w', encoding='utf-8') as config_file:
       config_file.writelines(json.dumps(device.to_config_json(), indent=4))
@@ -333,10 +336,8 @@ def save_device(self, device: Device):
     """Edit and save an existing device config."""
 
     # Obtain the config file path
-    config_file_path = os.path.join(self._root_dir,
-                                      LOCAL_DEVICES_DIR,
-                                      device.device_folder,
-                                      DEVICE_CONFIG)
+    config_file_path = os.path.join(self._root_dir, LOCAL_DEVICES_DIR,
+                                    device.device_folder, DEVICE_CONFIG)
 
     with open(config_file_path, 'w+', encoding='utf-8') as config_file:
       config_file.writelines(json.dumps(device.to_config_json(), indent=4))
@@ -349,9 +350,8 @@ def save_device(self, device: Device):
   def delete_device(self, device: Device):
 
     # Obtain the config file path
-    device_folder = os.path.join(self._root_dir,
-                                  LOCAL_DEVICES_DIR,
-                                  device.device_folder)
+    device_folder = os.path.join(self._root_dir, LOCAL_DEVICES_DIR,
+                                 device.device_folder)
 
     # Delete the device directory
     shutil.rmtree(device_folder)
@@ -366,17 +366,13 @@ def start(self):
     self._start_network()
 
     self.get_net_orc().get_listener().register_callback(
-      self._device_discovered,
-      [NetworkEvent.DEVICE_DISCOVERED]
-    )
+        self._device_discovered, [NetworkEvent.DEVICE_DISCOVERED])
 
     if self._net_only:
       LOGGER.info('Network only option configured, no tests will be run')
     else:
       self.get_net_orc().get_listener().register_callback(
-          self._device_stable,
-          [NetworkEvent.DEVICE_STABLE]
-      )
+          self._device_stable, [NetworkEvent.DEVICE_STABLE])
 
     self.get_net_orc().start_listener()
     LOGGER.info('Waiting for devices on the network...')
@@ -511,16 +507,12 @@ def start_ui(self):
     client = docker.from_env()
 
     try:
-      client.containers.run(
-            image='testrun/ui',
-            auto_remove=True,
-            name='tr-ui',
-            hostname='testrun.io',
-            detach=True,
-            ports={
-              '80': 8080
-            }
-      )
+      client.containers.run(image='testrun/ui',
+                            auto_remove=True,
+                            name='tr-ui',
+                            hostname='testrun.io',
+                            detach=True,
+                            ports={'80': 8080})
     except ImageNotFound as ie:
       LOGGER.error('An error occured whilst starting the UI. ' +
                    'Please investigate and try again.')
@@ -549,16 +541,14 @@ def start_ws(self):
     client = docker.from_env()
 
     try:
-      client.containers.run(
-            image='testrun/ws',
-            auto_remove=True,
-            name='tr-ws',
-            detach=True,
-            ports={
-              '9001': 9001,
-              '1883': 1883
-            }
-      )
+      client.containers.run(image='testrun/ws',
+                            auto_remove=True,
+                            name='tr-ws',
+                            detach=True,
+                            ports={
+                                '9001': 9001,
+                                '1883': 1883
+                            })
     except ImageNotFound as ie:
       LOGGER.error('An error occured whilst starting the websockets server. ' +
                    'Please investigate and try again.')

testing/baseline/test_baseline

@@ -40,7 +40,7 @@ sudo cp testing/baseline/system.json local/system.json
 # Copy device configs to testrun
 sudo cp -r testing/device_configs/* local/devices
 
-sudo bin/testrun --single-intf --no-ui --validate > $TESTRUN_OUT 2>&1 & 
+sudo bin/testrun --single-intf --no-ui --target 02:42:aa:00:01:01 -fw 1.0 --validate > $TESTRUN_OUT 2>&1 & 
 TPID=$!
 
 # Time to wait for testrun to be ready