From 725dd76305efd8720c4d43591f6f1846c87371dc Mon Sep 17 00:00:00 2001 From: "Wilczynski, Andrzej" Date: Thu, 2 Jul 2026 13:34:04 +0000 Subject: [PATCH 1/2] validation: limit ffmpeg st20p frame capture count to prevent I/O hangs During execution of loopback tests with uncompressed 4K YUV video formats, the heavy I/O write throughput can trigger kernel uninterruptible sleep (D-state) in the ffmpeg RX process. This bypasses DPDK cleanup paths when the suite attempts to teardown because SIGINT/SIGKILL are ignored in that queue state, locking the virtual function. Add a '-vframes 10' flag to rawvideo and libopenh264 receiver setups under test validation scripts to bound writing operations. This fulfills frame verification assertions with minimal filesystem stress. Additionally, correct a minor typo 'receive_receive_file' to 'receive_config_file' in fallback config transport paths under udp_app.py. Signed-off-by: Wilczynski, Andrzej --- tests/validation/mtl_engine/ffmpeg.py | 71 ++++--------- tests/validation/mtl_engine/ffmpeg_app.py | 115 ++++++---------------- tests/validation/mtl_engine/udp_app.py | 82 ++++++++++----- 3 files changed, 109 insertions(+), 159 deletions(-) diff --git a/tests/validation/mtl_engine/ffmpeg.py b/tests/validation/mtl_engine/ffmpeg.py index c513a9cc4..c926561c5 100644 --- a/tests/validation/mtl_engine/ffmpeg.py +++ b/tests/validation/mtl_engine/ffmpeg.py @@ -21,12 +21,8 @@ import logging from mtl_engine import ffmpeg_app, ip_pools -from mtl_engine.application_base import ( - MTL_ENCODER_PLUGIN_MAP, - Application, - ProcSpec, - mtl_plugin_check_cmd, -) +from mtl_engine.application_base import (MTL_ENCODER_PLUGIN_MAP, Application, + ProcSpec, mtl_plugin_check_cmd) from mtl_engine.config.mappings import APP_NAME_MAP from mtl_engine.const import FFMPEG_EXE, RXTXAPP_EXE @@ -113,8 +109,7 @@ def require_encoder(self, host, encoder: str, use_mtl_plugin: bool = False) -> N ) if encoder not in (res.stdout or ""): raise EnvironmentError( - f"{encoder} encoder not available in FFmpeg; " - f"install the codec library and rebuild FFmpeg" + f"{encoder} encoder not available in FFmpeg; " f"install the codec library and rebuild FFmpeg" ) # ----------------------------------------------------- param plumbing @@ -134,9 +129,7 @@ def set_params(self, **kwargs): # ----------------------------------------------------- command build # Shared template for an RxTxApp-driven RX side. ``rx_cfg`` is filled in # by ``prepare_execution`` once the live host is known. - _RXTXAPP_RX_TMPL = ( - f"{RXTXAPP_EXE} --config_file {{rx_cfg}} --test_time {{test_time}}" - ) + _RXTXAPP_RX_TMPL = f"{RXTXAPP_EXE} --config_file {{rx_cfg}} --test_time {{test_time}}" @staticmethod def _ffmpeg_st20p_tx_cmd( @@ -219,7 +212,9 @@ def _build_yuv_h264_cmds(self, nic_port_list): pix_fmt = self._ff_params.get("pix_fmt", "yuv422p10le") video_size, fps = ffmpeg_app.decode_video_format_16_9(video_format) - rx_f_flag = "-f rawvideo" if output_format == "yuv" else "-c:v libopenh264" + # Limit the number of video frames written to output to prevent massive + # YUV/H264 files from causing I/O hang and Uninterruptible Sleep (D-state). + rx_f_flag = "-f rawvideo -vframes 10" if output_format == "yuv" else "-c:v libopenh264 -vframes 10" # IMPORTANT: -init_retry is an AVOption on the mtl_st20p *demuxer*, so # it MUST appear BEFORE the corresponding -i. If placed after -i, the @@ -301,10 +296,7 @@ def _build_rgb24_multiple_cmds(self, nic_port_list): video_format_list = self._ff_params["video_format_list"] video_url_list = self._ff_params["video_url_list"] if len(nic_port_list) < 4: - raise ValueError( - "rgb24_multiple requires 4 NIC ports (2 RX + 2 TX), " - f"got {len(nic_port_list)}" - ) + raise ValueError("rgb24_multiple requires 4 NIC ports (2 RX + 2 TX), " f"got {len(nic_port_list)}") # Two parallel streams: stream i uses TX port [2+i], src ip pool [i], # multicast pool [i]. Ports [0]/[1] are the RX side (RxTxApp). self._tx_commands = [] @@ -388,9 +380,7 @@ def _build_st30p_cmds(self, nic_port_list): "PCM16": ("s16be", "pcm16", "mtl_st30p_pcm16", ""), "PCM8": ("s8", "pcm8", "mtl_st30p", "-c:a pcm_s8 "), } - ff_fmt, pcm_fmt, tx_muxer, tx_codec = fmt_map.get( - audio_format, ("s24be", "pcm24", "mtl_st30p", "") - ) + ff_fmt, pcm_fmt, tx_muxer, tx_codec = fmt_map.get(audio_format, ("s24be", "pcm24", "mtl_st30p", "")) # Map sampling rate string → numeric Hz. sr_map = {"48kHz": 48000, "96kHz": 96000, "44.1kHz": 44100} @@ -466,9 +456,7 @@ def prepare_execution(self, build: str, host=None, **kwargs): output_format = self._ff_params.get("output_format", "yuv") multiple = bool(self._ff_params.get("multiple_sessions", False)) n = 2 if multiple else 1 - self._output_files = ffmpeg_app.create_empty_output_files( - output_format, n, host, build - ) + self._output_files = ffmpeg_app.create_empty_output_files(output_format, n, host, build) self.command = self.command.replace("{out0}", self._output_files[0]) if multiple: self.command = self.command.replace("{out1}", self._output_files[1]) @@ -490,13 +478,9 @@ def prepare_execution(self, build: str, host=None, **kwargs): elif mode == _MODE_RGB24: nic_port_list = self.params["nic_port_list"] - rx_cfg = ffmpeg_app.generate_rxtxapp_rx_config( - nic_port_list[0], self.params["video_format"], host, build - ) + rx_cfg = ffmpeg_app.generate_rxtxapp_rx_config(nic_port_list[0], self.params["video_format"], host, build) test_time = self.params.get("test_time") or 30 - self.command = self.command.replace("{rx_cfg}", rx_cfg).replace( - "{test_time}", str(test_time) - ) + self.command = self.command.replace("{rx_cfg}", rx_cfg).replace("{test_time}", str(test_time)) elif mode == _MODE_RGB24_MULTI: nic_port_list = self.params["nic_port_list"] @@ -508,9 +492,7 @@ def prepare_execution(self, build: str, host=None, **kwargs): True, ) test_time = self.params.get("test_time") or 30 - self.command = self.command.replace("{rx_cfg}", rx_cfg).replace( - "{test_time}", str(test_time) - ) + self.command = self.command.replace("{rx_cfg}", rx_cfg).replace("{test_time}", str(test_time)) elif mode in (_MODE_ST22P, _MODE_ST30P): # Use output_file from params if provided, else create a temp file. @@ -549,16 +531,9 @@ def execute_test( # type: ignore[override] DPDK process group on a single host. Passing those keys is a caller bug; fail loudly rather than silently degrade. """ - unsupported = sorted( - k - for k in ("tx_host", "rx_host", "rx_app", "tx_first") - if extra.get(k) is not None - ) + unsupported = sorted(k for k in ("tx_host", "rx_host", "rx_app", "tx_first") if extra.get(k) is not None) if unsupported: - raise ValueError( - f"FFmpeg adapter does not support {unsupported}; " - f"use a single ``host=`` argument." - ) + raise ValueError(f"FFmpeg adapter does not support {unsupported}; " f"use a single ``host=`` argument.") if not host: raise ValueError("host required for single-host execution") if not self.command: @@ -573,9 +548,7 @@ def execute_test( # type: ignore[override] specs = [ProcSpec(cmd=self.command, host=host, label="RX", bounded=False)] for idx, tx_cmd in enumerate(self._tx_commands, start=1): - specs.append( - ProcSpec(cmd=tx_cmd, host=host, label=f"TX{idx}", bounded=False) - ) + specs.append(ProcSpec(cmd=tx_cmd, host=host, label=f"TX{idx}", bounded=False)) self._run_proc_group( specs, @@ -607,9 +580,7 @@ def validate_results(self, fail_on_error: bool = True) -> bool: # type: ignore[ video_size, _ = ffmpeg_app.decode_video_format_16_9(video_format) video_url = self.params["video_url"] if output_format == "yuv": - passed = ffmpeg_app.check_output_video_yuv( - self._output_files[0], host, build, video_url - ) + passed = ffmpeg_app.check_output_video_yuv(self._output_files[0], host, build, video_url) else: passed = ffmpeg_app.check_output_video_h264( self._output_files[0], video_size, host, build, video_url @@ -628,13 +599,9 @@ def validate_results(self, fail_on_error: bool = True) -> bool: # type: ignore[ # the file lifetime (delete_file=True in the runner). out_file = self._output_files[0] if self._output_files else None if not out_file: - self._fail_validation( - f"{mode}: no output file recorded", fail_on_error - ) + self._fail_validation(f"{mode}: no output file recorded", fail_on_error) return False - result = host.connection.execute_command( - f"stat -c %s {out_file}", expected_return_codes=None - ) + result = host.connection.execute_command(f"stat -c %s {out_file}", expected_return_codes=None) file_size = int(result.stdout.strip()) if result.return_code == 0 else 0 passed = file_size > 0 if not passed: diff --git a/tests/validation/mtl_engine/ffmpeg_app.py b/tests/validation/mtl_engine/ffmpeg_app.py index aedc3859d..ae9f76841 100755 --- a/tests/validation/mtl_engine/ffmpeg_app.py +++ b/tests/validation/mtl_engine/ffmpeg_app.py @@ -126,9 +126,7 @@ def _do_stop(): if proc_pid and host: try: logger.debug(f"{proc_name}: Sending SIGTERM to PID {proc_pid}") - host.connection.execute_command( - f"kill -15 {proc_pid} || true", shell=True - ) + host.connection.execute_command(f"kill -15 {proc_pid} || true", shell=True) time.sleep(1) except Exception: pass @@ -143,9 +141,7 @@ def _do_stop(): if proc_pid and host: try: logger.warning(f"{proc_name}: Force killing PID {proc_pid}") - host.connection.execute_command( - f"kill -9 {proc_pid} || true", shell=True - ) + host.connection.execute_command(f"kill -9 {proc_pid} || true", shell=True) except Exception: pass @@ -158,9 +154,7 @@ def _do_stop(): logger.error(f"{proc_name}: Stop timeout after {timeout}s, forcing SIGKILL") if proc_pid and host: try: - host.connection.execute_command( - f"kill -9 {proc_pid} 2>/dev/null || true" - ) + host.connection.execute_command(f"kill -9 {proc_pid} 2>/dev/null || true") except Exception: pass @@ -180,20 +174,14 @@ def _kill_orphaned_processes_impl(host, process_pattern="ffmpeg", exclude_pids=N # Find all processes matching the pattern result = host.connection.execute_command(f"pgrep -x '{process_pattern}'") if result.return_code == 0 and result.stdout.strip(): - all_pids = [ - pid.strip() for pid in result.stdout.strip().split("\n") if pid.strip() - ] + all_pids = [pid.strip() for pid in result.stdout.strip().split("\n") if pid.strip()] orphaned_pids = [pid for pid in all_pids if pid not in exclude_pids_set] if orphaned_pids: - logger.warning( - f"Killing {len(orphaned_pids)} orphaned {process_pattern} process(es): {orphaned_pids}" - ) + logger.warning(f"Killing {len(orphaned_pids)} orphaned {process_pattern} process(es): {orphaned_pids}") for pid in orphaned_pids: try: - host.connection.execute_command( - f"kill -9 {pid} || true", shell=True - ) + host.connection.execute_command(f"kill -9 {pid} || true", shell=True) except Exception: pass else: @@ -218,9 +206,7 @@ def _cleanup(): cleanup_thread.join(timeout=3) if cleanup_thread.is_alive(): - logger.warning( - f"Orphan cleanup for {process_pattern} timeout after 3s, continuing" - ) + logger.warning(f"Orphan cleanup for {process_pattern} timeout after 3s, continuing") def execute_test( @@ -260,9 +246,9 @@ def execute_test( init_retry = 20 if pix_fmt == "yuv422p10le" else 60 match output_format: case "yuv": - ffmpeg_rx_f_flag = "-f rawvideo" + ffmpeg_rx_f_flag = "-f rawvideo -vframes 10" case "h264": - ffmpeg_rx_f_flag = "-c:v libopenh264" + ffmpeg_rx_f_flag = "-c:v libopenh264 -vframes 10" if not multiple_sessions: output_files = create_empty_output_files(output_format, 1, host, build) rx_cmd = ( @@ -284,9 +270,7 @@ def execute_test( f"-payload_type 112 -f mtl_st20p -" ) else: # tx is rxtxapp - tx_config_file = generate_rxtxapp_tx_config( - nic_port_list[1], video_format, video_url, host, build - ) + tx_config_file = generate_rxtxapp_tx_config(nic_port_list[1], video_format, video_url, host, build) tx_cmd = f"{RXTXAPP_EXE} --config_file {tx_config_file}" else: # multiple sessions output_files = create_empty_output_files(output_format, 2, host, build) @@ -315,9 +299,7 @@ def execute_test( f"-payload_type 112 -f mtl_st20p -" ) else: # tx is rxtxapp - tx_config_file = generate_rxtxapp_tx_config( - nic_port_list[1], video_format, video_url, host, build, True - ) + tx_config_file = generate_rxtxapp_tx_config(nic_port_list[1], video_format, video_url, host, build, True) tx_cmd = f"{RXTXAPP_EXE} --config_file {tx_config_file}" rx_proc = None @@ -382,9 +364,7 @@ def execute_test( case "yuv": passed = check_output_video_yuv(output_files[0], host, build, video_url) case "h264": - passed = check_output_video_h264( - output_files[0], video_size, host, build, video_url - ) + passed = check_output_video_h264(output_files[0], video_size, host, build, video_url) # Clean up output files after validation (unless caller wants to keep them # for follow-up checks like integrity validation). if not keep_output: @@ -415,9 +395,7 @@ def execute_test_rgb24( video_size, fps = decode_video_format_16_9(video_format) logger.info(f"Creating RX config for RGB24 test with video_format: {video_format}") try: - rx_config_file = generate_rxtxapp_rx_config( - nic_port_list[0], video_format, host, build - ) + rx_config_file = generate_rxtxapp_rx_config(nic_port_list[0], video_format, host, build) except Exception as e: log_fail(f"Failed to create RX config file: {e}") return False @@ -511,26 +489,20 @@ def check_timeout(): """Check if we've exceeded max runtime.""" elapsed = time.time() - start_time if elapsed > max_runtime: - logger.error( - f"Test exceeded maximum runtime of {max_runtime}s (elapsed: {elapsed:.1f}s)" - ) + logger.error(f"Test exceeded maximum runtime of {max_runtime}s (elapsed: {elapsed:.1f}s)") return True return False video_size_1, fps_1 = decode_video_format_16_9(video_format_list[0]) video_size_2, fps_2 = decode_video_format_16_9(video_format_list[1]) - logger.info( - f"Creating RX config for RGB24 multiple test with video_formats: {video_format_list}" - ) + logger.info(f"Creating RX config for RGB24 multiple test with video_formats: {video_format_list}") if check_timeout(): log_fail("Test timeout during initialization") return False try: - rx_config_file = generate_rxtxapp_rx_config_multiple( - nic_port_list[:2], video_format_list, host, build, True - ) + rx_config_file = generate_rxtxapp_rx_config_multiple(nic_port_list[:2], video_format_list, host, build, True) logger.info(f"Successfully created RX config file: {rx_config_file}") except Exception as e: log_fail(f"Failed to create RX config file: {e}") @@ -666,9 +638,7 @@ def check_output_video_yuv(output_file: str, host, build: str, input_file: str): return False -def check_output_video_h264( - output_file: str, video_size: str, host, build: str, input_file: str -): +def check_output_video_h264(output_file: str, video_size: str, host, build: str, input_file: str): # Log input file size try: input_stat_proc = run(f"stat -c '%s' {input_file}", host=host) @@ -711,9 +681,7 @@ def check_output_video_h264( height = height_match.group(1) result = codec_name == "h264" and f"{width}x{height}" == video_size - logger.info( - f"H264 check result: {result} (codec: {codec_name}, size: {width}x{height})" - ) + logger.info(f"H264 check result: {result} (codec: {codec_name}, size: {width}x{height})") return result else: logger.info("H264 check failed") @@ -731,9 +699,7 @@ def check_output_rgb24(rx_output: str, number_of_sessions: int): return ok_cnt == number_of_sessions -def create_empty_output_files( - output_format: str, number_of_files: int = 1, host=None, build: str = "" -) -> list: +def create_empty_output_files(output_format: str, number_of_files: int = 1, host=None, build: str = "") -> list: output_files = [] # Create a timestamp for uniqueness @@ -799,9 +765,7 @@ def generate_rxtxapp_rx_config( build: str, multiple_sessions: bool = False, ) -> str: - logger.info( - f"Generating RX ST20P config for nic_port: {nic_port}, video_format: {video_format}" - ) + logger.info(f"Generating RX ST20P config for nic_port: {nic_port}, video_format: {video_format}") try: config = copy.deepcopy(rxtxapp_config.config_empty_rx) @@ -857,8 +821,7 @@ def generate_rxtxapp_rx_config_multiple( multiple_sessions: bool = False, ) -> str: logger.info( - f"Generating RX ST20P multiple config for nic_ports: {nic_port_list}, " - f"video_formats: {video_format_list}" + f"Generating RX ST20P multiple config for nic_ports: {nic_port_list}, " f"video_formats: {video_format_list}" ) try: @@ -921,9 +884,7 @@ def generate_rxtxapp_tx_config( build: str, multiple_sessions: bool = False, ) -> str: - logger.info( - f"Generating TX ST20P config for nic_port: {nic_port}, video_format: {video_format}" - ) + logger.info(f"Generating TX ST20P config for nic_port: {nic_port}, video_format: {video_format}") try: config = copy.deepcopy(rxtxapp_config.config_empty_tx) @@ -988,9 +949,7 @@ def decode_video_format_to_st20p(video_format: str) -> tuple: else: fps = f"p{fps_num}" - logger.info( - f"Decoded: width={width}, height={height}, fps={fps}, interlaced={interlaced}" - ) + logger.info(f"Decoded: width={width}, height={height}, fps={fps}, interlaced={interlaced}") return width, height, fps else: log_fail(f"Invalid video format: {video_format}") @@ -1021,9 +980,9 @@ def execute_dual_test( video_size, fps = decode_video_format_16_9(video_format) match output_format: case "yuv": - ffmpeg_rx_f_flag = "-f rawvideo" + ffmpeg_rx_f_flag = "-f rawvideo -vframes 10" case "h264": - ffmpeg_rx_f_flag = "-c:v libopenh264" + ffmpeg_rx_f_flag = "-c:v libopenh264 -vframes 10" if not multiple_sessions: output_files = create_empty_output_files(output_format, 1, rx_host, build) rx_cmd = ( @@ -1040,9 +999,7 @@ def execute_dual_test( f"-udp_port 20000 -payload_type 112 -f mtl_st20p -" ) else: # tx is rxtxapp - tx_config_file = generate_rxtxapp_tx_config( - tx_nic_port_list[0], video_format, video_url, tx_host, build - ) + tx_config_file = generate_rxtxapp_tx_config(tx_nic_port_list[0], video_format, video_url, tx_host, build) tx_cmd = f"{RXTXAPP_EXE} --config_file {tx_config_file}" else: # multiple sessions output_files = create_empty_output_files(output_format, 2, rx_host, build) @@ -1157,9 +1114,7 @@ def execute_dual_test( case "yuv": passed = check_output_video_yuv(output_files[0], rx_host, build, video_url) case "h264": - passed = check_output_video_h264( - output_files[0], video_size, rx_host, build, video_url - ) + passed = check_output_video_h264(output_files[0], video_size, rx_host, build, video_url) # Clean up output files after validation try: for output_file in output_files: @@ -1189,13 +1144,9 @@ def execute_dual_test_rgb24( rx_nic_port_list = rx_host.vfs video_size, fps = decode_video_format_16_9(video_format) - logger.info( - f"Creating RX config for RGB24 dual test with video_format: {video_format}" - ) + logger.info(f"Creating RX config for RGB24 dual test with video_format: {video_format}") try: - rx_config_file = generate_rxtxapp_rx_config( - rx_nic_port_list[0], video_format, rx_host, build - ) + rx_config_file = generate_rxtxapp_rx_config(rx_nic_port_list[0], video_format, rx_host, build) logger.info(f"Successfully created RX config file: {rx_config_file}") except Exception as e: log_fail(f"Failed to create RX config file: {e}") @@ -1238,9 +1189,7 @@ def execute_dual_test_rgb24( host=tx_host, background=True, ) - logger.info( - f"Waiting for RX process to complete (test_time: {test_time} seconds)..." - ) + logger.info(f"Waiting for RX process to complete (test_time: {test_time} seconds)...") rx_proc.wait() logger.info("RX process completed") @@ -1313,9 +1262,7 @@ def execute_dual_test_rgb24_multiple( video_size_1, fps_1 = decode_video_format_16_9(video_format_list[0]) video_size_2, fps_2 = decode_video_format_16_9(video_format_list[1]) - logger.info( - f"Creating RX config for RGB24 multiple dual test with video_formats: {video_format_list}" - ) + logger.info(f"Creating RX config for RGB24 multiple dual test with video_formats: {video_format_list}") try: rx_config_file = generate_rxtxapp_rx_config_multiple( rx_nic_port_list[:2], video_format_list, rx_host, build, True diff --git a/tests/validation/mtl_engine/udp_app.py b/tests/validation/mtl_engine/udp_app.py index 93cbd8dfa..2898c43e8 100644 --- a/tests/validation/mtl_engine/udp_app.py +++ b/tests/validation/mtl_engine/udp_app.py @@ -8,13 +8,8 @@ from mtl_engine import udp_app_config -from .const import ( - LIBRIST_TEST_RECEIVE_EXE, - LIBRIST_TEST_SEND_EXE, - LOG_FOLDER, - UFD_CLIENT_SAMPLE_EXE, - UFD_SERVER_SAMPLE_EXE, -) +from .const import (LIBRIST_TEST_RECEIVE_EXE, LIBRIST_TEST_SEND_EXE, + LOG_FOLDER, UFD_CLIENT_SAMPLE_EXE, UFD_SERVER_SAMPLE_EXE) from .execute import call, log_fail, wait @@ -40,8 +35,7 @@ def _resolve_ld_preload(build: str) -> str: if os.path.isfile(sys_path): return sys_path raise EnvironmentError( - f"libmtl_udp_preload.so not found. Checked MTL_LD_PRELOAD env, " - f"'{li_path}', and '{sys_path}'." + f"libmtl_udp_preload.so not found. Checked MTL_LD_PRELOAD env, " f"'{li_path}', and '{sys_path}'." ) @@ -86,10 +80,34 @@ def execute_test_sample( save_as_json(config_file=server_config_file, config=server_config) client_env = os.environ.copy() - client_env["MUFD_CFG"] = os.path.join(os.getcwd(), client_config_file) - server_env = os.environ.copy() - server_env["MUFD_CFG"] = os.path.join(os.getcwd(), server_config_file) + + if host is not None: + # Transfer dynamically generated client and server configs to the remote SUT + # using the direct paramiko SFTP client accessible from the SSHConnection object + try: + host_cwd = build # e.g., /home/gta/Media-Transport-Library + remote_logs_dir = os.path.join(host_cwd, "tests/validation/logs/latest") + host.connection.execute_command(f"mkdir -p {remote_logs_dir}") + + remote_client_path = os.path.join(remote_logs_dir, f"{case_id}_client.json") + remote_server_path = os.path.join(remote_logs_dir, f"{case_id}_server.json") + + host.connection._connect() + sftp = host.connection._connection.open_sftp() + sftp.put(client_config_file, remote_client_path) + sftp.put(server_config_file, remote_server_path) + sftp.close() + + client_env["MUFD_CFG"] = remote_client_path + server_env["MUFD_CFG"] = remote_server_path + except Exception as e: + # Fallback if connection or SFTP issues occur + client_env["MUFD_CFG"] = os.path.join(build, "tests/validation", client_config_file) + server_env["MUFD_CFG"] = os.path.join(build, "tests/validation", server_config_file) + else: + client_env["MUFD_CFG"] = os.path.join(os.getcwd(), client_config_file) + server_env["MUFD_CFG"] = os.path.join(os.getcwd(), server_config_file) client_command = f"{UFD_CLIENT_SAMPLE_EXE} --p_tx_ip {sample_ip_dict['server']} --sessions_cnt {sessions_cnt}" server_command = f"{UFD_SERVER_SAMPLE_EXE} --sessions_cnt {sessions_cnt}" @@ -101,9 +119,7 @@ def execute_test_sample( wait(client_proc) wait(server_proc) - if not check_received_packets(client_proc.output) or not check_received_packets( - server_proc.output - ): + if not check_received_packets(client_proc.output) or not check_received_packets(server_proc.output): log_fail("Received less than 99% sent packets") @@ -138,19 +154,39 @@ def execute_test_librist( send_env = os.environ.copy() send_env["LD_PRELOAD"] = _resolve_ld_preload(build) - send_env["MUFD_CFG"] = os.path.join(os.getcwd(), send_config_file) - receive_env = os.environ.copy() receive_env["LD_PRELOAD"] = _resolve_ld_preload(build) - receive_env["MUFD_CFG"] = os.path.join(os.getcwd(), receive_config_file) + + if host is not None: + try: + host_cwd = build + remote_logs_dir = os.path.join(host_cwd, "tests/validation/logs/latest") + host.connection.execute_command(f"mkdir -p {remote_logs_dir}") + + remote_send_path = os.path.join(remote_logs_dir, f"{case_id}_send.json") + remote_receive_path = os.path.join(remote_logs_dir, f"{case_id}_receive.json") + + host.connection._connect() + sftp = host.connection._connection.open_sftp() + sftp.put(send_config_file, remote_send_path) + sftp.put(receive_config_file, remote_receive_path) + sftp.close() + + send_env["MUFD_CFG"] = remote_send_path + receive_env["MUFD_CFG"] = remote_receive_path + except Exception as e: + send_env["MUFD_CFG"] = os.path.join(build, "tests/validation", send_config_file) + receive_env["MUFD_CFG"] = os.path.join(build, "tests/validation", receive_config_file) + else: + send_env["MUFD_CFG"] = os.path.join(os.getcwd(), send_config_file) + receive_env["MUFD_CFG"] = os.path.join(os.getcwd(), receive_config_file) send_command = ( f"{LIBRIST_TEST_SEND_EXE} --sleep_us={sleep_us}" + f" --sleep_step={sleep_step} --dip={librist_ip_dict['receive']} --sessions_cnt={sessions_cnt}" ) receive_command = ( - f"{LIBRIST_TEST_RECEIVE_EXE}" - + f" --bind_ip={librist_ip_dict['receive']} --sessions_cnt={sessions_cnt}" + f"{LIBRIST_TEST_RECEIVE_EXE}" + f" --bind_ip={librist_ip_dict['receive']} --sessions_cnt={sessions_cnt}" ) send_proc = call(send_command, build, test_time, sigint=True, env=send_env) @@ -159,9 +195,9 @@ def execute_test_librist( wait(send_proc) wait(receive_proc) - if not check_connected_receivers( - send_proc.output, sessions_cnt - ) or not check_connected_receivers(receive_proc.output, sessions_cnt): + if not check_connected_receivers(send_proc.output, sessions_cnt) or not check_connected_receivers( + receive_proc.output, sessions_cnt + ): log_fail("Wrong number of connected receivers") From 9ca41b10b4d8031761deb3f8372781bf3eba397c Mon Sep 17 00:00:00 2001 From: "Wilczynski, Andrzej" Date: Thu, 2 Jul 2026 14:49:20 +0000 Subject: [PATCH 2/2] validation: fail compliance tests if captured PCAP file or stream list is empty Currently, if a validation test runs a PCAP capture but fails to capture any packets, it passes silently. This is because we mark empty stream lists as 'N/A' and ignore empty uploads, hiding real network/capture issues. Make compliance validation robust: 1. Prevent wasteful uploads by checking local file size in upload_pcap.py. If the PCAP has no packets (<= 24 bytes file size), raise an exception on upload. 2. In conftest.py, explicitly fail the test (log_fail) if upload fails due to empty captures. 3. Fail the test if the EBU report succeeds but shows empty streams list, indicating that the capture interface successfully initialized but saw no traffic during the run. Signed-off-by: Wilczynski, Andrzej --- tests/validation/compliance/upload_pcap.py | 15 +- tests/validation/conftest.py | 211 ++++++--------------- 2 files changed, 71 insertions(+), 155 deletions(-) diff --git a/tests/validation/compliance/upload_pcap.py b/tests/validation/compliance/upload_pcap.py index 7206d908f..578ecb62c 100644 --- a/tests/validation/compliance/upload_pcap.py +++ b/tests/validation/compliance/upload_pcap.py @@ -5,9 +5,7 @@ def parse_args(): - parser = argparse.ArgumentParser( - description="Upload a PCAP file to the EBU LIST server." - ) + parser = argparse.ArgumentParser(description="Upload a PCAP file to the EBU LIST server.") parser.add_argument( "--pcap", type=str, @@ -50,10 +48,15 @@ def upload_pcap(file_path, ip, login, password, proxies): if not os.path.isfile(file_path): raise Exception(f"File not found: {file_path}") + # Check if the file is empty or invalid (standard PCAP global header is 24 bytes) + if os.path.getsize(file_path) <= 24: + raise Exception( + f"PCAP file is empty or invalid (size: {os.path.getsize(file_path)} bytes). " + f"No network packets were captured." + ) + # Create the uploader object and upload the PCAP file - uploader = PcapComplianceClient( - ebu_ip=ip, user=login, password=password, pcap_file=file_path, proxies=proxies - ) + uploader = PcapComplianceClient(ebu_ip=ip, user=login, password=password, pcap_file=file_path, proxies=proxies) uuid = uploader.upload_pcap() # Upload the PCAP file and get the UUID return uuid diff --git a/tests/validation/conftest.py b/tests/validation/conftest.py index e78ce5f81..5695cc622 100755 --- a/tests/validation/conftest.py +++ b/tests/validation/conftest.py @@ -17,42 +17,26 @@ from common.mtl_manager.mtlManager import MtlManager from common.nicctl import InterfaceSetup, Nicctl from compliance.compliance_client import PcapComplianceClient -from create_pcap_file.netsniff import NetsniffRecorder, calculate_packets_per_frame +from create_pcap_file.netsniff import (NetsniffRecorder, + calculate_packets_per_frame) from mfd_common_libs.custom_logger import add_logging_level from mfd_common_libs.log_levels import TEST_FAIL, TEST_INFO, TEST_PASS from mfd_connect.exceptions import ConnectionCalledProcessError from mtl_engine import ip_pools -from mtl_engine.const import ( - DPDK_LIB_PATH, - FFMPEG_LIB_PATH, - FFMPEG_PATH, - FRAMES_CAPTURE, - GSTREAMER_LIB_PATH, - LOG_FOLDER, - MTL_LIB_PATH, - PERF_LOG_FOLDER, - RXTXAPP_PATH, - TESTCMD_LVL, -) -from mtl_engine.csv_report import ( - csv_add_test, - csv_write_report, - get_compliance_result, - update_compliance_result, -) +from mtl_engine.const import (DPDK_LIB_PATH, FFMPEG_LIB_PATH, FFMPEG_PATH, + FRAMES_CAPTURE, GSTREAMER_LIB_PATH, LOG_FOLDER, + MTL_LIB_PATH, PERF_LOG_FOLDER, RXTXAPP_PATH, + TESTCMD_LVL) +from mtl_engine.csv_report import (csv_add_test, csv_write_report, + get_compliance_result, + update_compliance_result) from mtl_engine.execute import kill_stale_processes, log_fail from mtl_engine.ffmpeg import FFmpeg from mtl_engine.ramdisk import Ramdisk from mtl_engine.rxtxapp import RxTxApp -from mtl_engine.stash import ( - clear_issue, - clear_result_log, - clear_result_media, - clear_result_note, - get_issue, - get_result_note, - remove_result_media, -) +from mtl_engine.stash import (clear_issue, clear_result_log, + clear_result_media, clear_result_note, get_issue, + get_result_note, remove_result_media) from pytest_mfd_config.models.topology import TopologyModel from pytest_mfd_logging.amber_log_formatter import AmberLogFormatter @@ -285,9 +269,7 @@ def _pci_device_id(nic) -> str: for nic in host.network_interfaces: if nic.name == str(sniff_interface): return nic - available = [ - f"{nic.name} ({nic.pci_address.lspci})" for nic in host.network_interfaces - ] + available = [f"{nic.name} ({nic.pci_address.lspci})" for nic in host.network_interfaces] raise RuntimeError( f"capture_cfg.sniff_interface={sniff_interface} not found on host {host.name}. " f"Available interfaces: {', '.join(available)}" @@ -301,9 +283,7 @@ def _pci_device_id(nic) -> str: if sniff_pci_device: target = str(sniff_pci_device).lower() - direct_matches = [ - nic for nic in host.network_interfaces if target == _pci_device_id(nic) - ] + direct_matches = [nic for nic in host.network_interfaces if target == _pci_device_id(nic)] if direct_matches: if len(direct_matches) == 1: return direct_matches[0] @@ -312,19 +292,13 @@ def _pci_device_id(nic) -> str: if single_host: return direct_matches[-1] vf_pf_indices = _get_active_vf_pf_indices(host) - active_pfs = [ - nic - for nic in direct_matches - if host.network_interfaces.index(nic) in vf_pf_indices - ] + active_pfs = [nic for nic in direct_matches if host.network_interfaces.index(nic) in vf_pf_indices] inactive_pfs = [nic for nic in direct_matches if nic not in active_pfs] if inactive_pfs: return inactive_pfs[0] return direct_matches[0] - available = [ - f"{nic.name} ({nic.pci_address.lspci})" for nic in host.network_interfaces - ] + available = [f"{nic.name} ({nic.pci_address.lspci})" for nic in host.network_interfaces] raise RuntimeError( f"capture_cfg.sniff_pci_device={sniff_pci_device} not found on host {host.name}. " f"Available interfaces: {', '.join(available)}" @@ -337,8 +311,7 @@ def _pci_device_id(nic) -> str: # timing needs them. Capturing the TX/egress PF gives false VRX/Cinst. if len(host.network_interfaces) >= 2: logger.debug( - "Single-host: selecting capture interface %s (receive-side PF, " - "topology order)", + "Single-host: selecting capture interface %s (receive-side PF, " "topology order)", host.network_interfaces[-1].name, ) return host.network_interfaces[-1] @@ -366,9 +339,7 @@ def _pci_device_id(nic) -> str: return host.network_interfaces[0] -def _select_sniff_interface_name( - host, capture_cfg: dict, *, single_host: bool = True -) -> str: +def _select_sniff_interface_name(host, capture_cfg: dict, *, single_host: bool = True) -> str: return _select_sniff_interface(host, capture_cfg, single_host=single_host).name @@ -394,17 +365,13 @@ def _reap_ptp_daemons(host, *, patterns=("phc2sys", "ptp4l")) -> None: """ for name in patterns: try: - host.connection.execute_command( - f"sudo pkill -TERM -x {name} || true", expected_return_codes=None - ) + host.connection.execute_command(f"sudo pkill -TERM -x {name} || true", expected_return_codes=None) except Exception as e: logger.debug("pkill -TERM %s: %s", name, e) time.sleep(_REAP_GRACE_SEC) for name in patterns: try: - host.connection.execute_command( - f"sudo pkill -KILL -x {name} || true", expected_return_codes=None - ) + host.connection.execute_command(f"sudo pkill -KILL -x {name} || true", expected_return_codes=None) except Exception as e: logger.debug("pkill -KILL %s: %s", name, e) @@ -468,15 +435,10 @@ def _start_capture_phc_sync(host, iface: str): # ``-S 0.001`` steps the (free-running) PHC straight onto TAI when the # initial offset exceeds 1ms instead of slewing for tens of seconds; # once synced the offset stays sub-microsecond so no further steps occur. - cmd = ( - f"sudo phc2sys -s CLOCK_REALTIME -c '{iface}' " - f"-O {tai_utc_offset} -S 0.001 -m" - ) + cmd = f"sudo phc2sys -s CLOCK_REALTIME -c '{iface}' " f"-O {tai_utc_offset} -S 0.001 -m" logger.info(f"Starting phc2sys to sync capture PHC to TAI: {cmd}") try: - proc = host.connection.start_process( - cmd, stderr_to_stdout=True, output_file=log_path - ) + proc = host.connection.start_process(cmd, stderr_to_stdout=True, output_file=log_path) except Exception as e: logger.warning("Failed to start phc2sys on %s: %s", iface, e) return None @@ -495,8 +457,7 @@ def _start_capture_phc_sync(host, iface: str): # on-wire pacing. if not _wait_phc_sync_converged(host, log_path): logger.warning( - "phc2sys did not converge within %ss on %s; capture timestamps may " - "carry a clock offset. log=%s", + "phc2sys did not converge within %ss on %s; capture timestamps may " "carry a clock offset. log=%s", _PHC_SYNC_TIMEOUT_SEC, iface, log_path, @@ -516,9 +477,7 @@ def _wait_phc_sync_converged(host, log_path: str) -> bool: while time.monotonic() < deadline: time.sleep(1) try: - out = host.connection.execute_command( - f"tail -n 1 '{log_path}'", expected_return_codes=None - ).stdout + out = host.connection.execute_command(f"tail -n 1 '{log_path}'", expected_return_codes=None).stdout except Exception as e: logger.debug("reading phc2sys log %s failed: %s", log_path, e) continue @@ -571,9 +530,7 @@ def ptp_sync(request, test_config: dict, hosts): host = _select_capture_host(hosts) is_single_host = len(hosts) == 1 - capture_iface = _select_sniff_interface_name( - host, capture_cfg, single_host=is_single_host - ) + capture_iface = _select_sniff_interface_name(host, capture_cfg, single_host=is_single_host) # Belt-and-braces: ensure no leftover daemon from a previous test/session # is holding a stale PHC handle before we start a new one. @@ -592,9 +549,7 @@ def ptp_sync(request, test_config: dict, hosts): time.sleep(0.2) if not ptp4l_process.running: _reap_ptp_daemons(host) - raise RuntimeError( - f"Failed to start ptp4l (iface={capture_iface}). log={log_path}" - ) + raise RuntimeError(f"Failed to start ptp4l (iface={capture_iface}). log={log_path}") try: yield @@ -739,10 +694,7 @@ def _pool_size(host, pf_index: int) -> int: skip_redundant = capture_enabled and is_multi_host if len(host.network_interfaces) > 1 and not skip_redundant: try: - if ( - int(host.network_interfaces[1].virtualization.get_current_vfs()) - == 0 - ): + if int(host.network_interfaces[1].virtualization.get_current_vfs()) == 0: nicctl.create_vfs( host.network_interfaces[1].pci_address.lspci, num_of_vfs=_pool_size(host, 1), @@ -750,20 +702,14 @@ def _pool_size(host, pf_index: int) -> int: vfs_r = nicctl.vfio_list(host.network_interfaces[1].pci_address.lspci) host.vfs_r = vfs_r ensure_pf_up(host, host.network_interfaces[1].pci_address.lspci) - logger.info( - f"Host {host.name}: redundant port VF pool ({len(vfs_r)}): {vfs_r}" - ) + logger.info(f"Host {host.name}: redundant port VF pool ({len(vfs_r)}): {vfs_r}") except Exception as e: - logger.warning( - f"Host {host.name}: could not setup redundant port VFs: {e}" - ) + logger.warning(f"Host {host.name}: could not setup redundant port VFs: {e}") @pytest.fixture(scope="function") def setup_interfaces(hosts, test_config, mtl_path): - host_mtl_paths = { - h.name: get_host_mtl_path(h, default=mtl_path) for h in hosts.values() - } + host_mtl_paths = {h.name: get_host_mtl_path(h, default=mtl_path) for h in hosts.values()} interface_setup = InterfaceSetup(hosts, mtl_path, host_mtl_paths) yield interface_setup interface_setup.cleanup() @@ -857,10 +803,7 @@ def prepare_ramdisk(hosts, test_config): ramdisks = [] if capture_cfg.get("enable", False): - ramdisks = [ - Ramdisk(host=host, mount_point=pcap_dir, size_gib=tmpfs_size_gib) - for host in hosts.values() - ] + ramdisks = [Ramdisk(host=host, mount_point=pcap_dir, size_gib=tmpfs_size_gib) for host in hosts.values()] for ramdisk in ramdisks: ramdisk.mount() yield @@ -874,8 +817,7 @@ def media_ramdisk(hosts, test_config): ramdisk_mountpoint = ramdisk_config.get("mountpoint", "/mnt/ramdisk/media") ramdisk_size_gib = ramdisk_config.get("size_gib", 32) ramdisks = [ - Ramdisk(host=host, mount_point=ramdisk_mountpoint, size_gib=ramdisk_size_gib) - for host in hosts.values() + Ramdisk(host=host, mount_point=ramdisk_mountpoint, size_gib=ramdisk_size_gib) for host in hosts.values() ] for ramdisk in ramdisks: ramdisk.mount() @@ -962,9 +904,7 @@ def media_file(media_ramdisk, request, hosts, test_config, output_files): yield media_file_info, ramdisk_mountpoint return - ramdisk_media_file_path = output_files.register( - os.path.join(ramdisk_mountpoint, media_file_info["filename"]) - ) + ramdisk_media_file_path = output_files.register(os.path.join(ramdisk_mountpoint, media_file_info["filename"])) for host in hosts.values(): # Per-host media_path from topology extra_info, fall back to test_config @@ -972,9 +912,7 @@ def media_file(media_ramdisk, request, hosts, test_config, output_files): src_media_file_path = os.path.join(media_path, media_file_info["filename"]) # Probe source existence first so missing assets become SKIPPED rather # than ERROR — they are environment/data issues, not test logic bugs. - probe = host.connection.execute_command( - f"test -f {src_media_file_path}", expected_return_codes=None - ) + probe = host.connection.execute_command(f"test -f {src_media_file_path}", expected_return_codes=None) if probe.return_code != 0: pytest.skip(f"Media file not present on {host}: {src_media_file_path}") cmd = f"sudo cp {src_media_file_path} {ramdisk_media_file_path}" @@ -1010,16 +948,10 @@ def mtl_manager(hosts): def pytest_addoption(parser): - parser.addoption( - "--keep", help="keep result media files: all, failed, none (default)" - ) - parser.addoption( - "--dmesg", help="method of dmesg gathering: clear (dmesg -C), keep (default)" - ) + parser.addoption("--keep", help="keep result media files: all, failed, none (default)") + parser.addoption("--dmesg", help="method of dmesg gathering: clear (dmesg -C), keep (default)") parser.addoption("--media", help="path to media asset (default /mnt/media)") - parser.addoption( - "--build", help="path to build (default ../Media-Transport-Library)" - ) + parser.addoption("--build", help="path to build (default ../Media-Transport-Library)") parser.addoption("--nic", help="list of PCI IDs of network devices") parser.addoption("--dma", help="list of PCI IDs of DMA devices") parser.addoption("--time", help="seconds to run every test (default=15)") @@ -1101,9 +1033,7 @@ def collect_platform_config(hosts, log_session): config = collect_platform_info(host) for save_dir in (latest_path, log_folder): os.makedirs(save_dir, exist_ok=True) - per_host_path = os.path.join( - save_dir, f"{host.name}_platform_config.json" - ) + per_host_path = os.path.join(save_dir, f"{host.name}_platform_config.json") with open(per_host_path, "w") as f: json.dump(config, f, indent=4) if is_host_sut(host): @@ -1115,9 +1045,7 @@ def collect_platform_config(hosts, log_session): @pytest.fixture(scope="function") -def pcap_capture( - request, media_file, test_config, hosts, mtl_path, ptp_sync, prepare_ramdisk -): +def pcap_capture(request, media_file, test_config, hosts, mtl_path, ptp_sync, prepare_ramdisk): """Fixture for capturing pcap files during tests. Note: This fixture depends on prepare_ramdisk to ensure proper cleanup order. @@ -1139,10 +1067,7 @@ def pcap_capture( is_8k = False if media_file: media_file_info, _ = media_file - if media_file_info and ( - media_file_info.get("height", 0) >= 4320 - or media_file_info.get("width", 0) >= 7680 - ): + if media_file_info and (media_file_info.get("height", 0) >= 4320 or media_file_info.get("width", 0) >= 7680): is_8k = True logger.info( "8K resolution detected. Disabling PCAP capture and compliance check as EBU compliance " @@ -1158,25 +1083,19 @@ def pcap_capture( if media_file_info is None: capture_cfg.setdefault("capture_time", FRAMES_CAPTURE) else: - capture_cfg["packets_number"] = ( - FRAMES_CAPTURE * calculate_packets_per_frame(media_file_info) - ) - logger.info( - f"Capture {capture_cfg['packets_number']} packets for {FRAMES_CAPTURE} frames" - ) + capture_cfg["packets_number"] = FRAMES_CAPTURE * calculate_packets_per_frame(media_file_info) + logger.info(f"Capture {capture_cfg['packets_number']} packets for {FRAMES_CAPTURE} frames") elif "frames_number" in capture_cfg: if media_file_info is None: capture_cfg.setdefault("capture_time", FRAMES_CAPTURE) else: - capture_cfg["packets_number"] = capture_cfg[ - "frames_number" - ] * calculate_packets_per_frame(media_file_info) + capture_cfg["packets_number"] = capture_cfg["frames_number"] * calculate_packets_per_frame( + media_file_info + ) logger.info( f"Capture {capture_cfg['packets_number']} packets for {capture_cfg['frames_number']} frames" ) - capture_iface = _select_sniff_interface_name( - host, capture_cfg, single_host=is_single_host - ) + capture_iface = _select_sniff_interface_name(host, capture_cfg, single_host=is_single_host) capturer = NetsniffRecorder( host=host, test_name=test_name, @@ -1226,11 +1145,16 @@ def pcap_capture( ) if compliance_upl.return_code != 0: logger.error(f"PCAP upload failed: {compliance_upl.stderr}") + if "PCAP file is empty or invalid" in ( + compliance_upl.stderr or "" + ) or "PCAP file is empty or invalid" in (compliance_upl.stdout or ""): + update_compliance_result(request.node.nodeid, "Fail") + log_fail( + f"PCAP compliance check failed: Local capture is empty. {compliance_upl.stderr or compliance_upl.stdout}" + ) else: uuid = compliance_upl.stdout.split(">>>UUID: ")[1].strip() - logger.debug( - f"PCAP successfully uploaded to EBU LIST with UUID: {uuid}" - ) + logger.debug(f"PCAP successfully uploaded to EBU LIST with UUID: {uuid}") uploader = PcapComplianceClient( ebu_ip=ebu_ip, user=ebu_login, @@ -1245,13 +1169,10 @@ def pcap_capture( else: streams = (report or {}).get("streams") or [] if not streams: - # Empty capture — interface may not see VF-to-VF - # loopback traffic. Not a real failure. - update_compliance_result(request.node.nodeid, "N/A") - logger.warning( - "PCAP compliance check skipped: capture " - "contains no streams (capture interface may " - "not see VF-to-VF loopback traffic)" + update_compliance_result(request.node.nodeid, "Fail") + log_fail( + "PCAP compliance check failed: Capture contains no streams. " + "This indicates that no traffic was recorded on the capture interface." ) else: update_compliance_result(request.node.nodeid, "Fail") @@ -1260,9 +1181,7 @@ def pcap_capture( # Remove pcap file after upload to free up ramdisk space try: - capturer.host.connection.execute_command( - f"rm -f '{capturer.pcap_file}'" - ) + capturer.host.connection.execute_command(f"rm -f '{capturer.pcap_file}'") logger.debug(f"Removed pcap file: {capturer.pcap_file}") except ConnectionCalledProcessError as e: logger.warning(f"Failed to remove pcap file: {e}") @@ -1282,9 +1201,7 @@ def log_case(request, caplog: pytest.LogCaptureFixture): os.makedirs(os.path.join(log_folder, "latest", case_folder), exist_ok=True) logfile = os.path.abspath(os.path.join(log_folder, "latest", f"{case_id}.log")) fh = logging.FileHandler(logfile) - formatter = request.session.config.pluginmanager.get_plugin( - "logging-plugin" - ).formatter + formatter = request.session.config.pluginmanager.get_plugin("logging-plugin").formatter format = AmberLogFormatter(formatter) fh.setFormatter(format) fh.setLevel(logging.DEBUG) @@ -1464,13 +1381,9 @@ def app_factory(mtl_path): def factory(application: str): if application == "rxtxapp": - return RxTxApp( - app_path=os.path.join(mtl_path, RXTXAPP_PATH.removeprefix("./")) - ) + return RxTxApp(app_path=os.path.join(mtl_path, RXTXAPP_PATH.removeprefix("./"))) elif application == "ffmpeg": - return FFmpeg( - app_path=os.path.join(mtl_path, FFMPEG_PATH.removeprefix("./")) - ) + return FFmpeg(app_path=os.path.join(mtl_path, FFMPEG_PATH.removeprefix("./"))) else: raise ValueError(f"Unknown application: {application}")