tools/psoc6/run_test_plan.py: WIP. FIXUP.

Signed-off-by: jaenrig-ifx <enriquezgarcia.external@infineon.com>

Author: jaenrig-ifx

tools/psoc6/run_test_plan.py

@@ -1,5 +1,6 @@
 import argparse
 from dataclasses import dataclass, field
+from abc import ABC, abstractmethod
 from typing import ClassVar
 from enum import Enum
 import logging
@@ -429,16 +430,17 @@ class TestPlanLogger:
 
     decorator_line_len = 41
 
-    def __init__(self):
-        pass
+    # def __init__(self):
+    #     pass
 
-    def test_plan_info(self, board, test_plan_file, hil_devs_file):
+    def test_plan_info(self, test_plan_file, hil_devs_file=None, board=None):
         print(
             f"{TestPlanLogger.blue_on}"
             + "#" * TestPlanLogger.decorator_line_len
             + f"{TestPlanLogger.color_off}"
         )
-        print(f"{TestPlanLogger.blue_on}> board        : {board}{TestPlanLogger.color_off}")
+        if board:
+            print(f"{TestPlanLogger.blue_on}> board        : {board}{TestPlanLogger.color_off}")
         print(f"test plan file : {os.path.relpath(test_plan_file)}")
         if hil_devs_file:
             print(f"hil devs file  : {os.path.relpath(hil_devs_file)}")
@@ -534,32 +536,21 @@ def test_summary_info(self, pass_test_name_list, fail_test_name_list, skip_test_
         )
 
 
-class TestPlanRunner:
-    def __init__(
-        self, test_plan_file, hil_devs_file, max_retries=0, dut_port=None, stub_port=None
-    ):
+class TestPlanRunner(ABC):
+    def __init__(self, test_plan_file):
         self.test_plan_file = test_plan_file
-        self.hil_devs_file = hil_devs_file
-        self.max_retries = max_retries
         self.logger = TestPlanLogger()
 
-        self.dut_port = dut_port
-        self.stub_port = stub_port
-        self.direct_mode = True if (self.dut_port is not None) else False
-        print("Direct mode:", self.direct_mode)
-
-    def run(self, board, test_name_list=[]):
+    def run(self, test_name_list=[], max_retries=0):
         test_list = self.get_test_list(test_name_list)
 
-        self.logger.test_plan_info(board, self.test_plan_file, self.hil_devs_file)
-
         pass_test_name_list = []
         skip_test_name_list = []
         fail_test_name_list = []
         pending_retries = True
         while pending_retries:
             for test in test_list:
-                dut_port, stub_port = self.get_test_device_ports(test, board)
+                dut_port, stub_port = self.get_test_device_ports(test)
 
                 if not test.are_required_devs_available(dut_port, stub_port):
                     skip_test_name_list.append(test.name)
@@ -569,11 +560,10 @@ def run(self, board, test_name_list=[]):
                 # TODO: Add device.switch management.
 
                 self.logger.test_info(test.name, dut_port, stub_port)
-
                 ret_code = test.run(dut_port, stub_port)
 
                 if ret_code != 0:
-                    TestRunnerRetries.update_upon_fail(test.name, self.max_retries)
+                    TestRunnerRetries.update_upon_fail(test.name, max_retries)
                     self.logger.test_fail_info(test.name)
                 else:
                     TestRunnerRetries.update_upon_success(test.name)
@@ -594,34 +584,49 @@ def run(self, board, test_name_list=[]):
         if fail_test_name_list:
             sys.exit(1)
 
-    def get_test_device_ports(self, test, board):
-        def get_ports_for_role(test, board, device_role):
-            allowed_board_list = test.get_allowed_board_list(device_role, board)
+    def get_test_list(self, test_name_list=[]):
+        test_plan_list = TestRunner.load_list_from_yaml(self.test_plan_file)
 
-            logging.debug(
-                f'ALLOWED {device_role} board list for test "{test.name}": {allowed_board_list}'
-            )
+        if test_name_list == []:
+            return test_plan_list
 
-            port_list = []
-            for device in allowed_board_list:
-                available_ports = get_devices_port(
-                    device.get("board"), self.hil_devs_file, device.get("version", None)
-                )
-                port_list.extend(available_ports)
+        test_list = []
+        for test_name in test_name_list:
+            for test in test_plan_list:
+                if test.name == test_name:
+                    test_list.append(test)
+
+        logging.debug(f"Selected test list: {[test.name for test in test_list]}")
+
+        return test_list
+
+    @abstractmethod
+    def get_test_device_ports(self, test):
+        pass
+
+
+class TestPlanRunnerHIL(TestPlanRunner):
+    def __init__(self, test_plan_file, hil_devs_file, board):
+        super().__init__(test_plan_file)
+        self.hil_devs_file = hil_devs_file
+        self.board = board
 
-            logging.debug(f'AVAILABLE {device_role} ports for test "{test.name}": {port_list}')
+    def set_board(self, board):
+        self.board = board
 
-            return port_list
+    def run(self, test_name_list=[], max_retries=0):
+        self.logger.test_plan_info(self.test_plan_file, self.hil_devs_file, self.board)
+        return super().run(test_name_list, max_retries)
 
+    """ 
+    Private methods
+    """
+
+    def get_test_device_ports(self, test):
         dut_port = None
         stub_port = None
 
-        if self.direct_mode:
-            dut_port = self.dut_port
-            stub_port = self.stub_port
-            return dut_port, stub_port
-
-        dut_port_list = get_ports_for_role(test, board, TestRunner.DeviceRole.DUT)
+        dut_port_list = self.get_ports_for_role(test, self.board, TestRunner.DeviceRole.DUT)
 
         if not dut_port_list:
             return dut_port, stub_port
@@ -630,37 +635,52 @@ def get_ports_for_role(test, board, device_role):
         dut_port = dut_port_list[0]
 
         if test.requires_multiple_devs():
-            stub_port_list = get_ports_for_role(test, board, TestRunner.DeviceRole.STUB)
+            stub_port_list = self.get_ports_for_role(test, self.board, TestRunner.DeviceRole.STUB)
 
             # Take any element from stub_port_list that is not dut_port
             for port in stub_port_list:
                 if port != dut_port:
                     stub_port = port
                     break
 
-            if not stub_port:
-                return dut_port, stub_port
-
         return dut_port, stub_port
 
-    def get_test_list(self, test_name_list=[]):
-        test_plan_list = TestRunner.load_list_from_yaml(self.test_plan_file)
+    def get_ports_for_role(self, test, board, device_role):
+        allowed_board_list = test.get_allowed_board_list(device_role, board)
 
-        if test_name_list == []:
-            return test_plan_list
+        port_list = []
+        for device in allowed_board_list:
+            available_ports = get_devices_port(
+                device.get("board"), self.hil_devs_file, device.get("version", None)
+            )
+            port_list.extend(available_ports)
 
-        test_list = []
-        for test_name in test_name_list:
-            for test in test_plan_list:
-                if test.name == test_name:
-                    test_list.append(test)
+        return port_list
 
-        logging.debug(f"Selected test list: {[test.name for test in test_list]}")
 
-        return test_list
+class TestPlanRunnerPorts(TestPlanRunner):
+    def __init__(self, test_plan_file, dut_port, stub_port):
+        super().__init__(test_plan_file)
+        self.dut_port = dut_port
+        self.stub_port = stub_port
+
+    def set_ports(self, dut_port, stub_port):
+        self.dut_port = dut_port
+        self.stub_port = stub_port
 
+    def run(self, test_name_list=[], max_retries=0):
+        self.logger.test_plan_info(self.test_plan_file)
+        return super().run(test_name_list, max_retries)
 
-class TestPlanCLI:
+    """ 
+    Private methods
+    """
+
+    def get_test_device_ports(self, test):
+        return self.dut_port, self.stub_port
+
+
+class TestPlanRunnerCLI:
     def __init__(self):
         self.parser = argparse.ArgumentParser(description="MicroPython test suites runner.")
         self.parser.add_argument("test_suite", nargs="*", type=str, help="Test suite to run.")
@@ -747,13 +767,16 @@ def set_validate_args(self, args):
 
 
 if __name__ == "__main__":
-    test_plan_cli = TestPlanCLI()
-    cli_args = test_plan_cli.parse()
-    test_plan_runner = TestPlanRunner(
-        cli_args.test_plan,
-        cli_args.hil_devs,
-        cli_args.max_retries,
-        cli_args.dut_port,
-        cli_args.stub_port,
-    )
-    test_plan_runner.run(cli_args.board, cli_args.test_suite)
+    test_plan_runner_cli = TestPlanRunnerCLI()
+    tpr_args = test_plan_runner_cli.parse()
+
+    # HIL device file based mode
+    if tpr_args.hil_devs:
+        test_plan_runner = TestPlanRunnerHIL(tpr_args.test_plan, tpr_args.hil_devs, tpr_args.board)
+    # Direct port passing mode
+    elif tpr_args.dut_port:
+        test_plan_runner = TestPlanRunnerPorts(
+            tpr_args.test_plan, tpr_args.dut_port, tpr_args.stub_port
+        )
+
+    test_plan_runner.run(tpr_args.test_suite, tpr_args.max_retries)