Added refactored tests for single host, st20p category. Based on new OOP approach.

Signed-off-by: Wilczynski, Andrzej <[email protected]>

Author: awilczyns

tests/validation/mtl_engine/application_base.py

@@ -8,15 +8,7 @@
 
 from .config.app_mappings import DEFAULT_PAYLOAD_TYPE_CONFIG, DEFAULT_PORT_CONFIG
 from .config.universal_params import UNIVERSAL_PARAMS
-
-# Import execution utilities with fallback
-try:
-    from .execute import run
-    from .RxTxApp import prepare_tcpdump
-except ImportError:
-    # Fallback for direct execution
-    from execute import run
-    from RxTxApp import prepare_tcpdump
+from .execute import run
 
 logger = logging.getLogger(__name__)
 
@@ -130,45 +122,29 @@ def get_session_default_payload_type(self, session_type: str) -> int:
             "ancillary": DEFAULT_PAYLOAD_TYPE_CONFIG["ancillary_payload_type"],
             "fastmetadata": DEFAULT_PAYLOAD_TYPE_CONFIG["fastmetadata_payload_type"],
         }
-        return payload_map.get(
-            session_type, DEFAULT_PAYLOAD_TYPE_CONFIG["st20p_payload_type"]
-        )
+        return payload_map.get(session_type, DEFAULT_PAYLOAD_TYPE_CONFIG["st20p_payload_type"])
 
     def get_common_session_params(self, session_type: str) -> dict:
         """Get common session parameters used across all session types."""
         default_port = self.get_session_default_port(session_type)
         default_payload = self.get_session_default_payload_type(session_type)
 
         return {
-            "replicas": self.universal_params.get(
-                "replicas", UNIVERSAL_PARAMS["replicas"]
-            ),
-            "start_port": int(
-                self.universal_params.get("port")
-                if self.was_user_provided("port")
-                else default_port
-            ),
+            "replicas": self.universal_params.get("replicas", UNIVERSAL_PARAMS["replicas"]),
+            "start_port": int(self.universal_params.get("port") if self.was_user_provided("port") else default_port),
             "payload_type": (
-                self.universal_params.get("payload_type")
-                if self.was_user_provided("payload_type")
-                else default_payload
+                self.universal_params.get("payload_type") if self.was_user_provided("payload_type") else default_payload
             ),
         }
 
     def get_common_video_params(self) -> dict:
         """Get common video parameters used across video session types."""
         return {
             "width": int(self.universal_params.get("width", UNIVERSAL_PARAMS["width"])),
-            "height": int(
-                self.universal_params.get("height", UNIVERSAL_PARAMS["height"])
-            ),
-            "interlaced": self.universal_params.get(
-                "interlaced", UNIVERSAL_PARAMS["interlaced"]
-            ),
+            "height": int(self.universal_params.get("height", UNIVERSAL_PARAMS["height"])),
+            "interlaced": self.universal_params.get("interlaced", UNIVERSAL_PARAMS["interlaced"]),
             "device": self.universal_params.get("device", UNIVERSAL_PARAMS["device"]),
-            "enable_rtcp": self.universal_params.get(
-                "enable_rtcp", UNIVERSAL_PARAMS["enable_rtcp"]
-            ),
+            "enable_rtcp": self.universal_params.get("enable_rtcp", UNIVERSAL_PARAMS["enable_rtcp"]),
         }
 
     def execute_test(
@@ -210,35 +186,26 @@ def execute_test(
             cmd = self.add_timeout(self.command, test_time)
             logger.info(f"[single] Running {framework_name} command: {cmd}")
             # Optional tcpdump capture hook retained for RxTxApp compatibility
-            if (
-                capture_cfg
-                and capture_cfg.get("enable")
-                and "prepare_tcpdump" in globals()
-            ):
+            if capture_cfg and capture_cfg.get("enable") and "prepare_tcpdump" in globals():
                 try:
-                    prepare_tcpdump(capture_cfg, host)
+                    # prepare_tcpdump not yet implemented; left to change in the future
+                    # prepare_tcpdump(capture_cfg, host)
+                    pass
                 except Exception as e:
                     logger.warning(f"capture setup failed: {e}")
             proc = self.start_process(cmd, build, test_time, host)
             try:
-                proc.wait(
-                    timeout=(test_time or 0)
-                    + self.universal_params.get("process_timeout_buffer", 90)
-                )
+                proc.wait(timeout=(test_time or 0) + self.universal_params.get("process_timeout_buffer", 90))
             except Exception:
-                logger.warning(
-                    f"{framework_name} process wait timed out (continuing to capture output)"
-                )
+                logger.warning(f"{framework_name} process wait timed out (continuing to capture output)")
             self.last_output = self.capture_stdout(proc, framework_name)
             self.last_return_code = getattr(proc, "returncode", None)
             return self.validate_results()
 
         # Dual-host execution (tx self, rx rx_app)
         assert rx_app is not None
         if not rx_app.command:
-            raise RuntimeError(
-                "rx_app has no prepared command (call create_command first)"
-            )
+            raise RuntimeError("rx_app has no prepared command (call create_command first)")
         tx_cmd = self.add_timeout(self.command, test_time)
         rx_cmd = rx_app.add_timeout(rx_app.command, test_time)
         primary_first = tx_first
@@ -258,16 +225,12 @@ def execute_test(
         logger.info(f"[dual] Starting second: {second_label} -> {second_cmd}")
         second_proc = self.start_process(second_cmd, build, test_time, second_host)
         # Wait processes
-        total_timeout = (test_time or 0) + self.universal_params.get(
-            "process_timeout_buffer", 90
-        )
+        total_timeout = (test_time or 0) + self.universal_params.get("process_timeout_buffer", 90)
         for p, label in [(first_proc, first_label), (second_proc, second_label)]:
             try:
                 p.wait(timeout=total_timeout)
             except Exception:
-                logger.warning(
-                    f"Process {label} wait timeout; capturing partial output"
-                )
+                logger.warning(f"Process {label} wait timeout; capturing partial output")
         # Capture outputs
         if primary_first:
             self.last_output = self.capture_stdout(first_proc, first_label)
@@ -310,9 +273,7 @@ def add_timeout(self, command: str, test_time: int, grace: int = None) -> str:
             return f"timeout {effective_test_time + grace} {command}"
         return command
 
-    def start_and_capture(
-        self, command: str, build: str, test_time: int, host, process_name: str
-    ):
+    def start_and_capture(self, command: str, build: str, test_time: int, host, process_name: str):
         """Start a single process and capture its stdout safely."""
         process = self.start_process(command, build, test_time, host)
         output = self.capture_stdout(process, process_name)
@@ -362,9 +323,7 @@ def extract_framerate(self, framerate_str, default: int = None) -> int:
         try:
             return int(float(num))
         except ValueError:
-            logger.warning(
-                f"Could not parse framerate '{framerate_str}', defaulting to {default}"
-            )
+            logger.warning(f"Could not parse framerate '{framerate_str}', defaulting to {default}")
             return default
 
     # Legacy execute_* abstract methods removed; unified execute_test used instead.
@@ -382,9 +341,7 @@ def capture_stdout(self, process, process_name: str) -> str:
             # Remote process objects (from mfd_connect) expose stdout via 'stdout_text'
             if hasattr(process, "stdout_text") and process.stdout_text:
                 output = process.stdout_text
-                logger.debug(
-                    f"{process_name} output (captured stdout_text): {output[:200]}..."
-                )
+                logger.debug(f"{process_name} output (captured stdout_text): {output[:200]}...")
                 return output
             # Local fallback (subprocess) may expose .stdout already consumed elsewhere
             if hasattr(process, "stdout") and process.stdout:
@@ -394,9 +351,7 @@ def capture_stdout(self, process, process_name: str) -> str:
                         output = process.stdout.read()
                     else:
                         output = str(process.stdout)
-                    logger.debug(
-                        f"{process_name} output (captured stdout): {output[:200]}..."
-                    )
+                    logger.debug(f"{process_name} output (captured stdout): {output[:200]}...")
                     return output
                 except Exception:
                     pass