Cleanup and removed hardocded IP addresses

Signed-off-by: Wilczynski, Andrzej <andrzej.wilczynski@intel.com>

Author: awilczyns

tests/validation/mtl_engine/application_base.py

@@ -1,12 +1,12 @@
-# Base Application Class for Media Transport Library
-# Provides common interface for all media application frameworks
+"""Base application class providing unified interface for media framework adapters."""
 
 import logging
 import re
 import time
 from abc import ABC, abstractmethod
 
-from .config.app_mappings import DEFAULT_PAYLOAD_TYPE_CONFIG, DEFAULT_PORT_CONFIG
+from .config.app_mappings import (DEFAULT_PAYLOAD_TYPE_CONFIG,
+                                  DEFAULT_PORT_CONFIG)
 from .config.universal_params import UNIVERSAL_PARAMS
 from .execute import run
 
@@ -98,71 +98,6 @@ def was_user_provided(self, param_name: str) -> bool:
         """Check if a parameter was explicitly provided by the user."""
         return param_name in self._user_provided_params
 
-    def get_session_default_port(self, session_type: str) -> int:
-        """Get default port for a specific session type."""
-        port_map = {
-            "st20p": DEFAULT_PORT_CONFIG["st20p_port"],
-            "st22p": DEFAULT_PORT_CONFIG["st22p_port"],
-            "st30p": DEFAULT_PORT_CONFIG["st30p_port"],
-            "video": DEFAULT_PORT_CONFIG["video_port"],
-            "audio": DEFAULT_PORT_CONFIG["audio_port"],
-            "ancillary": DEFAULT_PORT_CONFIG["ancillary_port"],
-            "fastmetadata": DEFAULT_PORT_CONFIG["fastmetadata_port"],
-        }
-        return port_map.get(session_type, DEFAULT_PORT_CONFIG["st20p_port"])
-
-    def get_session_default_payload_type(self, session_type: str) -> int:
-        """Get default payload type for a specific session type."""
-        payload_map = {
-            "st20p": DEFAULT_PAYLOAD_TYPE_CONFIG["st20p_payload_type"],
-            "st22p": DEFAULT_PAYLOAD_TYPE_CONFIG["st22p_payload_type"],
-            "st30p": DEFAULT_PAYLOAD_TYPE_CONFIG["st30p_payload_type"],
-            "video": DEFAULT_PAYLOAD_TYPE_CONFIG["video_payload_type"],
-            "audio": DEFAULT_PAYLOAD_TYPE_CONFIG["audio_payload_type"],
-            "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"]
-        )
-
-    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
-            ),
-            "payload_type": (
-                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"]
-            ),
-            "device": self.universal_params.get("device", UNIVERSAL_PARAMS["device"]),
-            "enable_rtcp": self.universal_params.get(
-                "enable_rtcp", UNIVERSAL_PARAMS["enable_rtcp"]
-            ),
-        }
-
     def prepare_execution(self, build: str, host=None, **kwargs):
         """Hook method called before execution to perform framework-specific setup.
 
@@ -223,20 +158,7 @@ def execute_test(
         if not is_dual:
             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()
-            ):
-                try:
-                    # 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)
-            # Start netsniff capture if provided (RxTxApp-specific)
             if netsniff and hasattr(self, "_start_netsniff_capture"):
                 try:
                     self._start_netsniff_capture(netsniff)
@@ -252,7 +174,7 @@ def execute_test(
                     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)
+            self.last_return_code = proc.return_code
             return self.validate_results()
 
         # Dual-host execution (tx self, rx rx_app)
@@ -297,75 +219,48 @@ def execute_test(
         else:
             rx_app.last_output = rx_app.capture_stdout(first_proc, first_label)
             self.last_output = self.capture_stdout(second_proc, second_label)
-        self.last_return_code = getattr(first_proc, "returncode", None)
-        rx_app.last_return_code = getattr(second_proc, "returncode", None)
+        self.last_return_code = first_proc.return_code
+        rx_app.last_return_code = second_proc.return_code
         tx_ok = self.validate_results()
         rx_ok = rx_app.validate_results()
         return tx_ok and rx_ok
 
-    # -------------------------
-    # Common helper utilities
-    # -------------------------
     def add_timeout(self, command: str, test_time: int, grace: int = None) -> str:
-        """Wrap command with timeout if test_time provided (adds a grace period)."""
+        """Wrap command with timeout if test_time provided.
+
+        Args:
+            command: Shell command to wrap
+            test_time: Test duration in seconds
+            grace: Grace period to add (default from universal_params)
+
+        Returns:
+            Command wrapped with timeout
+        """
         if grace is None:
             grace = self.universal_params.get("timeout_grace", 10)
-        # If the command already has an internal --test_time X argument, ensure the wrapper
-        # timeout is >= that internal value + grace to avoid premature SIGTERM (RC 124).
+
+        # Extract internal --test_time to prevent premature timeout termination
         internal_test_time = None
         m = re.search(r"--test_time\s+(\d+)", command)
         if m:
             try:
                 internal_test_time = int(m.group(1))
             except ValueError:
                 internal_test_time = None
+
         effective_test_time = test_time or internal_test_time
         if internal_test_time and test_time and internal_test_time != test_time:
             logger.debug(
-                f"Mismatch between execute_test test_time={test_time} and command --test_time {internal_test_time}; "
-                f"using max"
+                f"Test time mismatch (execute={test_time}, command={internal_test_time}); using max"
             )
             effective_test_time = max(internal_test_time, test_time)
         elif internal_test_time and not test_time:
             effective_test_time = internal_test_time
+
         if effective_test_time and not command.strip().startswith("timeout "):
             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
-    ):
-        """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)
-        return process, output
-
-    def start_dual_with_delay(
-        self,
-        tx_command: str,
-        rx_command: str,
-        build: str,
-        test_time: int,
-        tx_host,
-        rx_host,
-        tx_first: bool,
-        sleep_interval: int,
-        tx_name: str,
-        rx_name: str,
-    ):
-        """Start two processes with an optional delay ordering TX/RX based on tx_first flag."""
-        if tx_first:
-            tx_process = self.start_process(tx_command, build, test_time, tx_host)
-            time.sleep(sleep_interval)
-            rx_process = self.start_process(rx_command, build, test_time, rx_host)
-        else:
-            rx_process = self.start_process(rx_command, build, test_time, rx_host)
-            time.sleep(sleep_interval)
-            tx_process = self.start_process(tx_command, build, test_time, tx_host)
-        tx_output = self.capture_stdout(tx_process, tx_name)
-        rx_output = self.capture_stdout(rx_process, rx_name)
-        return (tx_process, rx_process, tx_output, rx_output)
-
     def extract_framerate(self, framerate_str, default: int = None) -> int:
         """Extract numeric framerate from various string or numeric forms (e.g. 'p25', '60')."""
         if default is None:
@@ -389,47 +284,42 @@ def extract_framerate(self, framerate_str, default: int = None) -> int:
             )
             return default
 
-    # Legacy execute_* abstract methods removed; unified execute_test used instead.
-
     def start_process(self, command: str, build: str, test_time: int, host):
-        """Start a process on the specified host."""
+        """Start a process on the specified host using mfd_connect."""
         logger.info(f"Starting {self.get_framework_name()} process...")
         buffer_val = self.universal_params.get("process_timeout_buffer", 90)
         timeout = (test_time or 0) + buffer_val
         return run(command, host=host, cwd=build, timeout=timeout)
 
     def capture_stdout(self, process, process_name: str) -> str:
-        """Capture stdout from a process."""
+        """Capture stdout from mfd_connect process.
+
+        Note: Must be called after process.wait() completes, as stdout_text
+        is only available after the process finishes.
+
+        Args:
+            process: mfd_connect process object with stdout_text attribute
+            process_name: Name for logging purposes
+
+        Returns:
+            Process stdout as string, or empty string if unavailable
+        """
         try:
-            # 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]}..."
-                )
-                return output
-            # Local fallback (subprocess) may expose .stdout already consumed elsewhere
-            if hasattr(process, "stdout") and process.stdout:
-                try:
-                    # Attempt to read if it's a file-like object
-                    if hasattr(process.stdout, "read"):
-                        output = process.stdout.read()
-                    else:
-                        output = str(process.stdout)
-                    logger.debug(
-                        f"{process_name} output (captured stdout): {output[:200]}..."
-                    )
-                    return output
-                except Exception:
-                    pass
-            logger.warning(f"No stdout available for {process_name}")
+            output = process.stdout_text or ""
+            if output:
+                logger.debug(f"{process_name} output: {output[:200]}...")
+            return output
+        except AttributeError:
+            logger.error(
+                f"Process object missing stdout_text attribute for {process_name}"
+            )
             return ""
         except Exception as e:
-            logger.error(f"Error capturing {process_name} output: {e}")
+            logger.error(f"Error capturing output from {process_name}: {e}")
             return ""
 
     def get_case_id(self) -> str:
-        """Generate a case ID for logging/debugging purposes."""
+        """Generate test case identifier from call stack for logging."""
         try:
             import inspect
 

tests/validation/mtl_engine/config/app_mappings.py

@@ -27,16 +27,6 @@
     },
 }
 
-# Default network configuration values
-DEFAULT_NETWORK_CONFIG = {
-    "nic_port": None,
-    "unicast_tx_ip": "192.168.17.101",
-    "unicast_rx_ip": "192.168.17.102",
-    "multicast_tx_ip": "192.168.17.101",
-    "multicast_rx_ip": "192.168.17.102",
-    "multicast_destination_ip": "239.168.48.9",
-    "default_config_file": "config.json",
-}
 
 # Default port configuration by session type
 DEFAULT_PORT_CONFIG = {