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}") 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")