diff --git a/.github/scripts/setup_validation.sh b/.github/scripts/setup_validation.sh index 9ae0fa0ed..4f60cf313 100755 --- a/.github/scripts/setup_validation.sh +++ b/.github/scripts/setup_validation.sh @@ -269,7 +269,12 @@ stage_preflight() { warn "preflight: out-of-tree ice driver not loaded (path=$ice_path)" missing=1 fi - free_mb=$(awk '/HugePages_Free/ {print $2*2}' /proc/meminfo) + # Free hugepage memory in MiB, derived from the ACTUAL default page size + # (Hugepagesize in /proc/meminfo). The old `*2` assumed 2 MiB pages and + # misreported hosts booted with default_hugepagesz=1G (e.g. 32 free 1G + # pages -> 64 MiB instead of 32768 MiB). Backward compatible: on 2 MiB + # default hosts Hugepagesize=2048 kB so the factor is still 2. + free_mb=$(awk '/^HugePages_Free:/ {f=$2} /^Hugepagesize:/ {sz=$2} END {print f * (sz/1024)}' /proc/meminfo) if ((free_mb < 1024)); then warn "preflight: hugepages free is ${free_mb} MiB (<1024 MiB)" missing=1 diff --git a/tests/validation/conftest.py b/tests/validation/conftest.py index e78ce5f81..034ecd4ce 100755 --- a/tests/validation/conftest.py +++ b/tests/validation/conftest.py @@ -1471,6 +1471,12 @@ def factory(application: str): return FFmpeg( app_path=os.path.join(mtl_path, FFMPEG_PATH.removeprefix("./")) ) + elif application == "gstreamer": + from mtl_engine.gstreamer import GStreamer + + # app_path is vestigial for GStreamer (gst-launch-1.0 is on PATH); + # pass the MTL build dir to satisfy the base constructor. + return GStreamer(app_path=mtl_path) else: raise ValueError(f"Unknown application: {application}") diff --git a/tests/validation/mtl_engine/GstreamerApp.py b/tests/validation/mtl_engine/GstreamerApp.py index 887277ebb..5d6bddc66 100755 --- a/tests/validation/mtl_engine/GstreamerApp.py +++ b/tests/validation/mtl_engine/GstreamerApp.py @@ -206,6 +206,7 @@ def setup_gstreamer_st20p_tx_pipeline( tx_queues: int, tx_framebuff_num: int = None, tx_fps: int = None, + enable_ptp: bool = False, ): connection_params = create_connection_params( dev_port=nic_port_list, @@ -254,6 +255,9 @@ def setup_gstreamer_st20p_tx_pipeline( pipeline_command.extend(["mtl_st20p_tx", f"tx-queues={tx_queues}"]) + if enable_ptp: + pipeline_command.append("enable-ptp=true") + if tx_framebuff_num is not None: pipeline_command.append(f"tx-framebuff-num={tx_framebuff_num}") @@ -280,6 +284,7 @@ def setup_gstreamer_st20p_rx_pipeline( rx_queues: int, rx_framebuff_num: int = None, rx_fps: int = None, + enable_ptp: bool = False, ): connection_params = create_connection_params( dev_port=nic_port_list, @@ -305,6 +310,9 @@ def setup_gstreamer_st20p_rx_pipeline( f"rx-fps={framerate}", ] + if enable_ptp: + pipeline_command.append("enable-ptp=true") + for key, value in connection_params.items(): pipeline_command.append(f"{key}={value}") diff --git a/tests/validation/mtl_engine/gstreamer.py b/tests/validation/mtl_engine/gstreamer.py new file mode 100644 index 000000000..d7330170c --- /dev/null +++ b/tests/validation/mtl_engine/gstreamer.py @@ -0,0 +1,378 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2026 Intel Corporation +"""GStreamer framework adapter (unified Application model). + +Wraps the procedural builders and orchestrator in :mod:`mtl_engine.GstreamerApp`, +which stay the single source of truth for pipeline token lists and pass/fail +criteria. ``create_command`` materialises the TX/RX gst-launch token lists from +the same UNIVERSAL_PARAMS vocabulary RxTxApp/FFmpeg use (translating to +GStreamer caps internally); ``execute_test`` delegates the TX+RX run to +:func:`GstreamerApp.execute_test`. + +Execution is delegated because the procedural orchestrator runs a TX+RX pipeline +pair on a single host and owns md5 output comparison, which the base single-host +run loop (one ``self.command``) does not model. The adapter still honors the +shared ``netsniff=`` capture hook so one parametrized test can drive RxTxApp, +FFmpeg and GStreamer identically. The base PTP extension and SIGINT->SIGKILL +stop ladder are not reused: GstreamerApp owns process teardown. + +``app_path`` is the MTL build directory: ``gst-launch-1.0`` is resolved from +PATH and the builders hardcode it, so ``app_path`` is used only as the default +``build`` dir (``--gst-plugin-path``) when a test does not pass ``build=`` +explicitly. +""" + +from __future__ import annotations + +from mtl_engine import GstreamerApp, ip_pools +from mtl_engine.application_base import Application +from mtl_engine.config.mappings import APP_NAME_MAP + +_ST40P_REDUNDANT_UDP_PORT = 40001 + +# UNIVERSAL_PARAMS -> GStreamer caps translation. These let a GStreamer adapter +# be driven by the same ``create_command(**universal_params)`` call that +# RxTxApp/FFmpeg use, so one parametrized test can target all three frameworks. +_AUDIO_FORMAT_TO_GST = {"PCM8": "S8", "PCM16": "S16BE", "PCM24": "S24BE"} +_AUDIO_SAMPLING_TO_HZ = {"44.1kHz": 44100, "48kHz": 48000, "96kHz": 96000} +# Channel-layout label -> count. Mirrors the RxTxApp channel vocabulary; ``U0N`` +# is "N user channels". +_AUDIO_CHANNEL_LABEL_TO_COUNT = { + "M": 1, + "DM": 2, + "ST": 2, + "LtRt": 2, + "51": 6, + "71": 8, + "222": 24, + "SGRP": 4, + "U01": 1, + "U02": 2, +} + + +class GStreamer(Application): + """GStreamer framework adapter (single-host TX+RX orchestrator).""" + + # Parameters the GStreamer pipelines understand but which are not part of + # UNIVERSAL_PARAMS. Stripped in ``set_params`` into ``self._gst`` so the + # base-class validation does not reject them. ``audio_format`` is also a + # universal key but carries a GStreamer caps value here, so it is routed + # into ``self._gst`` rather than ``self.params``. + _GSTREAMER_ONLY_KEYS = ( + "build", + "gst_format", + "tx_queues", + "rx_queues", + "tx_payload_type", + "rx_payload_type", + "audio_format", + "audio_channels", + "audio_rate", + "redundant", + "frame_info_path", + "capture_metadata", + "skip_file_compare", + "rx_timeout", + "tx_framebuff_cnt", + "rx_framebuff_cnt", + "tx_fps", + "tx_did", + "tx_sdid", + "tx_rfc8331", + "tx_user_pacing", + "tx_user_controlled_pacing", + "tx_user_controlled_pacing_offset", + "tx_interlaced", + "rx_interlaced", + "tx_split_anc_by_pkt", + "tx_test_mode", + "tx_test_pkt_count", + "tx_test_pacing_ns", + "rx_disable_auto_detect", + "rx_rtp_ring_size", + ) + + def __init__(self, app_path, config_file_path=None): + super().__init__(app_path, config_file_path) + self._gst: dict = {} + self._tx_command: list[str] = [] + self._rx_command: list[str] = [] + self._input_file = None + self._output_file = None + self._build = None + self._last_passed = False + + def get_app_name(self) -> str: + return "GStreamer" + + def get_executable_name(self) -> str: + return APP_NAME_MAP["gstreamer"] + + def set_params(self, **kwargs): + """Strip GStreamer-only kwargs into ``self._gst`` first, then defer.""" + self._gst = {} + for key in self._GSTREAMER_ONLY_KEYS: + if key in kwargs: + self._gst[key] = kwargs.pop(key) + super().set_params(**kwargs) + + def _create_command_and_config(self) -> tuple: + # ``build`` may be supplied explicitly (legacy GStreamer-only tests) or + # derived from ``app_path`` (the MTL build dir that ``app_factory`` + # passes), so the same call used by RxTxApp/FFmpeg works unchanged. + self._build = self._gst.get("build") or self.app_path + if not self._build: + raise ValueError("GStreamer requires a build dir (pass build= or app_path)") + + nic_port_list = self.params["nic_port_list"] + if not nic_port_list: + raise ValueError("nic_port_list is required") + + session_type = self.params["session_type"] + if session_type == "st20p": + self._build_st20p(nic_port_list) + elif session_type in ("st30", "st30p"): + self._build_st30(nic_port_list) + elif session_type == "st40p": + self._build_st40p(nic_port_list) + else: + raise ValueError(f"Unsupported GStreamer session_type: {session_type}") + + self._input_file = self.params["input_file"] + self._output_file = self.params["output_file"] + return " ".join(self._rx_command), None + + # ------------------------------------------------ param translation + def _resolve_gst_video_format(self) -> str: + """GStreamer caps name from ``gst_format`` or universal transport format.""" + fmt = self._gst.get("gst_format") + if fmt: + return fmt + src = self.params.get("transport_format") or self.params.get("pixel_format") + return GstreamerApp.video_format_change(src) + + def _resolve_framerate(self) -> str: + """Numeric framerate string (``"p25"`` -> ``"25"``) for the gst builders.""" + return str(self.extract_framerate(self.params["framerate"])) + + def _resolve_audio_format(self) -> str: + """GStreamer audio caps (``S8``/``S16BE``/``S24BE``) from any vocabulary.""" + fmt = self._gst.get("audio_format") or self.params.get("audio_format") + return _AUDIO_FORMAT_TO_GST.get(fmt, fmt) + + def _resolve_audio_channels(self) -> int: + ch = self._gst.get("audio_channels", self.params.get("audio_channels")) + if isinstance(ch, (list, tuple)): + ch = ch[0] if ch else "U02" + if isinstance(ch, int): + return ch + if ch in _AUDIO_CHANNEL_LABEL_TO_COUNT: + return _AUDIO_CHANNEL_LABEL_TO_COUNT[ch] + return int(ch) + + def _resolve_audio_rate(self) -> int: + rate = self._gst.get("audio_rate") + if rate is not None: + return int(rate) + samp = self.params.get("audio_sampling") + if isinstance(samp, (int, float)): + return int(samp) + return _AUDIO_SAMPLING_TO_HZ.get(samp, 48000) + + def _build_st20p(self, nic_port_list): + gst_format = self._resolve_gst_video_format() + framerate = self._resolve_framerate() + enable_ptp = bool(self.params.get("enable_ptp", False)) + self._tx_command = GstreamerApp.setup_gstreamer_st20p_tx_pipeline( + build=self._build, + nic_port_list=nic_port_list[0], + input_path=self.params["input_file"], + width=self.params["width"], + height=self.params["height"], + framerate=framerate, + format=gst_format, + tx_payload_type=self._gst.get("tx_payload_type", 112), + tx_queues=self._gst.get("tx_queues", 4), + enable_ptp=enable_ptp, + ) + self._rx_command = GstreamerApp.setup_gstreamer_st20p_rx_pipeline( + build=self._build, + nic_port_list=nic_port_list[1], + output_path=self.params["output_file"], + width=self.params["width"], + height=self.params["height"], + framerate=framerate, + format=gst_format, + rx_payload_type=self._gst.get("rx_payload_type", 112), + rx_queues=self._gst.get("rx_queues", 4), + enable_ptp=enable_ptp, + ) + + def _build_st30(self, nic_port_list): + audio_format = self._resolve_audio_format() + channels = self._resolve_audio_channels() + sampling = self._resolve_audio_rate() + self._tx_command = GstreamerApp.setup_gstreamer_st30_tx_pipeline( + build=self._build, + nic_port_list=nic_port_list[0], + input_path=self.params["input_file"], + tx_payload_type=self._gst.get("tx_payload_type", 111), + tx_queues=self._gst.get("tx_queues", 4), + audio_format=audio_format, + channels=channels, + sampling=sampling, + ) + self._rx_command = GstreamerApp.setup_gstreamer_st30_rx_pipeline( + build=self._build, + nic_port_list=nic_port_list[1], + output_path=self.params["output_file"], + rx_payload_type=self._gst.get("rx_payload_type", 111), + rx_queues=self._gst.get("rx_queues", 4), + rx_audio_format=GstreamerApp.audio_format_change( + audio_format, rx_side=True + ), + rx_channels=channels, + rx_sampling=sampling, + ) + + def _build_st40p(self, nic_port_list): + tx_payload = self._gst.get("tx_payload_type", 113) + rx_payload = self._gst.get("rx_payload_type", 113) + queues = self._gst.get("tx_queues", 4) + rx_queues = self._gst.get("rx_queues", 4) + timeout = self._gst.get("rx_timeout", 15) + + tx_red = {} + rx_red = {} + rx_primary = nic_port_list[1] + if self._gst.get("redundant"): + if len(nic_port_list) < 4: + raise ValueError("st40p redundant requires at least 4 NIC ports") + if len(ip_pools.tx) < 2 or len(ip_pools.rx) < 2: + raise ValueError("st40p redundant requires at least 2 TX and 2 RX IPs") + rx_primary = nic_port_list[2] + tx_red = dict( + dev_port_red=nic_port_list[1], + dev_ip_red=ip_pools.rx[1], + ip_red=ip_pools.tx[1], + udp_port_red=_ST40P_REDUNDANT_UDP_PORT, + ) + rx_red = dict( + dev_port_red=nic_port_list[3], + dev_ip_red=ip_pools.tx[1], + ip_red=ip_pools.rx[1], + udp_port_red=_ST40P_REDUNDANT_UDP_PORT, + ) + + self._tx_command = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + build=self._build, + nic_port_list=nic_port_list[0], + input_path=self.params["input_file"], + tx_payload_type=tx_payload, + tx_queues=queues, + tx_framebuff_cnt=self._gst.get("tx_framebuff_cnt"), + tx_fps=self._gst.get("tx_fps"), + tx_did=self._gst.get("tx_did"), + tx_sdid=self._gst.get("tx_sdid"), + tx_rfc8331=self._gst.get("tx_rfc8331", False), + tx_user_pacing=self._gst.get("tx_user_pacing", False), + tx_user_controlled_pacing=self._gst.get("tx_user_controlled_pacing", False), + tx_user_controlled_pacing_offset=self._gst.get( + "tx_user_controlled_pacing_offset", 0 + ), + tx_interlaced=self._gst.get("tx_interlaced", False), + tx_split_anc_by_pkt=self._gst.get("tx_split_anc_by_pkt", False), + tx_test_mode=self._gst.get("tx_test_mode"), + tx_test_pkt_count=self._gst.get("tx_test_pkt_count", 0), + tx_test_pacing_ns=self._gst.get("tx_test_pacing_ns", 0), + **tx_red, + ) + self._rx_command = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( + build=self._build, + nic_port_list=rx_primary, + output_path=self.params["output_file"], + rx_payload_type=rx_payload, + rx_queues=rx_queues, + timeout=timeout, + capture_metadata=self._gst.get("capture_metadata", False), + rx_interlaced=self._gst.get("rx_interlaced", False), + rx_disable_auto_detect=self._gst.get("rx_disable_auto_detect", False), + rx_framebuff_cnt=self._gst.get("rx_framebuff_cnt"), + frame_info_path=self._gst.get("frame_info_path"), + rx_rtp_ring_size=self._gst.get("rx_rtp_ring_size"), + **rx_red, + ) + + def execute_test( # type: ignore[override] + self, + build: str, + test_time: int = 30, + host=None, + sleep_interval: int = 4, + tx_first: bool = False, + fail_on_error: bool = True, + netsniff=None, + interface_setup=None, + skip_file_compare=None, + log_frame_info: bool = True, + suppress_fail_logs: bool = False, + **extra, + ) -> bool: + """Delegate execution to :func:`GstreamerApp.execute_test`. + + The procedural orchestrator owns process lifetime and md5 output + comparison, so this does not reuse the base single-host run loop (which + drives only one ``self.command``; GStreamer needs a TX+RX pair on one + host). It still honors the common ``netsniff=`` capture hook so a single + parametrized test can drive RxTxApp, FFmpeg and GStreamer identically. + ``suppress_fail_logs`` is forwarded so negative/rejection tests can + silence the failure log dump they expect. + """ + if not (self._tx_command and self._rx_command): + raise RuntimeError("create_command() must be called first") + + if skip_file_compare is None: + skip_file_compare = self._gst.get("skip_file_compare", False) + + # Arm a bounded packet capture that overlaps the stream window. The + # capture runs in the background for ``test_time`` while GstreamerApp + # drives the TX/RX pipelines synchronously. With PTP enabled the plugin + # blocks streaming until the clock locks, so the capture window is + # extended by the PTP sync budget to overlap the steady-state stream. + if netsniff is not None: + ptp_budget = ( + self.params.get("ptp_sync_time", 50) + if self.params.get("enable_ptp", False) + else 0 + ) + self.params["test_time"] = test_time + ptp_budget + self._start_netsniff_capture(netsniff) + + passed = GstreamerApp.execute_test( + build=build, + tx_command=self._tx_command, + rx_command=self._rx_command, + input_file=self._input_file, + output_file=self._output_file, + test_time=test_time, + host=host, + sleep_interval=sleep_interval, + tx_first=tx_first, + suppress_fail_logs=suppress_fail_logs, + skip_file_compare=skip_file_compare, + log_frame_info=log_frame_info, + ) + self._last_passed = bool(passed) + if not passed: + if fail_on_error: + self._fail_validation("GStreamer test failed", fail_on_error) + return False + return passed + + def validate_results(self, fail_on_error: bool = True) -> bool: # type: ignore[override] + return self._last_passed + + def _resolve_capture_dst_ip(self): # type: ignore[override] + """Destination IP the GStreamer TX pipeline streams to (for netsniff).""" + return ip_pools.rx[0] if ip_pools.rx else None diff --git a/tests/validation/tests/single/gstreamer/__init__.py b/tests/validation/tests/single/gstreamer/__init__.py deleted file mode 100755 index e69de29bb..000000000 diff --git a/tests/validation/tests/single/gstreamer/audio_format/__init__.py b/tests/validation/tests/single/gstreamer/audio_format/__init__.py deleted file mode 100755 index e69de29bb..000000000 diff --git a/tests/validation/tests/single/gstreamer/audio_format/test_audio_format.py b/tests/validation/tests/single/gstreamer/audio_format/test_audio_format.py deleted file mode 100755 index 063a351c8..000000000 --- a/tests/validation/tests/single/gstreamer/audio_format/test_audio_format.py +++ /dev/null @@ -1,115 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright(c) 2024-2025 Intel Corporation - -"""GStreamer ST30 audio format validation. - -Runs ST30 pipelines over multiple PCM formats, channel counts, and sampling -rates to confirm end-to-end transport and caps negotiation. This focuses on -format support; no audio integrity check is performed. -""" - -import os - -import mtl_engine.media_creator as media_create -import pytest -from common.nicctl import InterfaceSetup -from mtl_engine import GstreamerApp - - -@pytest.mark.nightly -@pytest.mark.parametrize("audio_format", ["S8", "S16BE", "S24BE"]) -@pytest.mark.parametrize("audio_channel", [1, 2, 6, 8]) -@pytest.mark.parametrize("audio_rate", [44100, 48000, 96000]) -def test_audio_format( - hosts, - mtl_path, - media, - setup_interfaces: InterfaceSetup, - audio_format, - audio_channel, - audio_rate, - test_time, - test_config, - prepare_ramdisk, - media_file, -): - if audio_rate == 96000 and ( - audio_channel == 8 - and (audio_format == "S16BE" or audio_format == "S24BE") - or audio_channel == 6 - and audio_format == "S24BE" - ): - pytest.skip( - "Audio {fmt}/{ch} invalid pkt_len; skipped".format( - fmt=audio_format, - ch=audio_channel, - ), - ) - - media_file_info, media_file_path = media_file - if not media_file_path: - raise ValueError( - "ramdisk was not setup correctly for media_file fixture", - ) - - # Get the first host for remote execution - host = list(hosts.values())[0] - input_file_path = os.path.join(media_file_path, "input_test_audio.pcm") - output_file_path = os.path.join(media_file_path, "output_test_audio.pcm") - - interfaces_list = setup_interfaces.get_interfaces_list_single( - test_config.get("interface_type", "VF"), - ) - - # media_create.create_audio_file_sox( - # sample_rate=audio_rate, - # channels=audio_channel, - # bit_depth=GstreamerApp.audio_format_change(audio_format), - # duration=10, - # frequency=440, - # output_path=input_file_path, - # host=host, - # ) - - # Input path unused; pipeline generates audio internally - tx_config = GstreamerApp.setup_gstreamer_st30_tx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, - tx_payload_type=111, - tx_queues=4, - audio_format=audio_format, - channels=audio_channel, - sampling=audio_rate, - ) - - rx_config = GstreamerApp.setup_gstreamer_st30_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, - rx_payload_type=111, - rx_queues=4, - rx_audio_format=GstreamerApp.audio_format_change( - audio_format, - rx_side=True, - ), - rx_channels=audio_channel, - rx_sampling=audio_rate, - ) - - try: - GstreamerApp.execute_test( - build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, - test_time=test_time, - host=host, - tx_first=False, - sleep_interval=1, - ) - finally: - pass - # media_create.remove_file(input_file_path, host=host) - media_create.remove_file(output_file_path, host=host) diff --git a/tests/validation/tests/single/gstreamer/video_format/__init__.py b/tests/validation/tests/single/gstreamer/video_format/__init__.py deleted file mode 100755 index e69de29bb..000000000 diff --git a/tests/validation/tests/single/gstreamer/video_format/test_video_format.py b/tests/validation/tests/single/gstreamer/video_format/test_video_format.py deleted file mode 100755 index 78134be12..000000000 --- a/tests/validation/tests/single/gstreamer/video_format/test_video_format.py +++ /dev/null @@ -1,96 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright(c) 2024-2025 Intel Corporation - -"""GStreamer ST20P video format validation. - -Generates synthetic clips and exercises ST20P TX/RX pipelines across the -catalog of supported GStreamer video formats to ensure negotiation, pacing, -and capture work for each advertised format. -""" - -import mtl_engine.media_creator as media_create -import pytest -from common.nicctl import InterfaceSetup -from mtl_engine import GstreamerApp -from mtl_engine.media_files import gstreamer_formats - - -@pytest.mark.nightly -@pytest.mark.parametrize( - "media_file", - list(gstreamer_formats.values()), - indirect=["media_file"], - ids=list(gstreamer_formats.keys()), -) -@pytest.mark.parametrize("file", gstreamer_formats.keys()) -def test_video_format( - hosts, - mtl_path, - setup_interfaces: InterfaceSetup, - file, - test_time, - test_config, - media_file, - prepare_ramdisk, -): - video_file, media_file_path = media_file - - # Get the first host for remote execution - host = list(hosts.values())[0] - interfaces_list = setup_interfaces.get_interfaces_list_single( - test_config.get("interface_type", "VF") - ) - - media_dir = host.connection.path(media_file_path).parent - media_dir = str(media_dir) - output_file_path = host.connection.path(media_dir, "output_video.yuv") - input_file_path = media_create.create_video_file( - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=GstreamerApp.video_format_change(video_file["format"]), - media_path=media_dir, - duration=3, - host=host, - ) - - tx_config = GstreamerApp.setup_gstreamer_st20p_tx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=GstreamerApp.video_format_change(video_file["format"]), - tx_payload_type=112, - tx_queues=4, - ) - - rx_config = GstreamerApp.setup_gstreamer_st20p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=GstreamerApp.video_format_change(video_file["format"]), - rx_payload_type=112, - rx_queues=4, - ) - - try: - GstreamerApp.execute_test( - build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, - test_time=test_time, - host=host, - tx_first=False, - sleep_interval=4, - ) - finally: - # Remove the video file after the test - media_create.remove_file(input_file_path, host=host) - media_create.remove_file(output_file_path, host=host) diff --git a/tests/validation/tests/single/gstreamer/video_resolution/__init__.py b/tests/validation/tests/single/gstreamer/video_resolution/__init__.py deleted file mode 100755 index e69de29bb..000000000 diff --git a/tests/validation/tests/single/gstreamer/video_resolution/test_video_resolution.py b/tests/validation/tests/single/gstreamer/video_resolution/test_video_resolution.py deleted file mode 100755 index 9b218c24c..000000000 --- a/tests/validation/tests/single/gstreamer/video_resolution/test_video_resolution.py +++ /dev/null @@ -1,115 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright(c) 2024-2025 Intel Corporation - -"""GStreamer ST20P video resolution validation. - -Sweeps ST20P pipelines across the catalog of YUV sample resolutions, including -format fallbacks for widths not divisible by six, to verify configuration, -pacing, and basic receive stability at varied sizes. -""" - -import os - -import mtl_engine.media_creator as media_create -import pytest -from common.nicctl import InterfaceSetup -from mtl_engine import GstreamerApp -from mtl_engine.media_files import yuv_files -from tests.xfail import SDBQ1971_conversion_v210_720p_error - - -@pytest.mark.nightly -@pytest.mark.parametrize( - "media_file", - list(yuv_files.values()), - indirect=["media_file"], - ids=list(yuv_files.keys()), -) -def test_video_resolutions( - hosts, - mtl_path, - setup_interfaces: InterfaceSetup, - request, - test_time, - test_config, - prepare_ramdisk, - media_file, -): - video_file, media_file_path = media_file - - # The st20 plugin can only produce v210 when the width is divisible by 6 - # (pixel groups are 6 pixels wide). Fall back to I422_10LE otherwise so - # 1280-wide sources succeed instead of tripping the converter. - gst_format = ( - "v210" - if video_file["width"] % 6 == 0 - else GstreamerApp.video_format_change(video_file["format"]) - ) - - # Get the first host for remote execution - host = list(hosts.values())[0] - interfaces_list = setup_interfaces.get_interfaces_list_single( - test_config.get("interface_type", "VF") - ) - - if gst_format == "v210": - SDBQ1971_conversion_v210_720p_error( - video_format=gst_format, - resolution_height=video_file["height"], - request=request, - ) - - media_dir = host.connection.path(media_file_path).parent - media_dir = str(media_dir) - input_file_path = media_create.create_video_file( - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=gst_format, - media_path=media_dir, - duration=3, - host=host, - ) - - tx_config = GstreamerApp.setup_gstreamer_st20p_tx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=gst_format, - tx_payload_type=112, - tx_queues=4, - ) - - rx_config = GstreamerApp.setup_gstreamer_st20p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=host.connection.path(media_dir, "output_video.yuv"), - width=video_file["width"], - height=video_file["height"], - framerate=video_file["fps"], - format=gst_format, - rx_payload_type=112, - rx_queues=4, - ) - try: - GstreamerApp.execute_test( - build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=os.path.join(media_dir, "output_video.yuv"), - test_time=test_time, - host=host, - tx_first=False, - sleep_interval=2, - ) - finally: - # Remove the video file after the test - media_create.remove_file(input_file_path, host=host) - media_create.remove_file( - os.path.join(media_dir, "output_video.yuv"), - host=host, - ) diff --git a/tests/validation/tests/single/ptp/test_st20p_ptp_gstreamer.py b/tests/validation/tests/single/ptp/test_st20p_ptp_gstreamer.py new file mode 100644 index 000000000..afb524f5b --- /dev/null +++ b/tests/validation/tests/single/ptp/test_st20p_ptp_gstreamer.py @@ -0,0 +1,97 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2026 Intel Corporation +"""GStreamer ST20P onboard-PTP validation. + +Exercises the GStreamer plugin ``enable-ptp`` property (onboard PTP client) on +a single-host TX+RX ST20P pipeline. GStreamer drives synthetic planar media, so +this validates that the stream transports and md5-matches end to end with PTP +enabled; it is GStreamer-only because the ``enable-ptp`` knob lives in the +plugin (the RxTxApp PTP coverage lives under the integrity-based ptp tests). +""" + +import os + +import mtl_engine.media_creator as media_create +import pytest +from common.nicctl import InterfaceSetup +from mtl_engine import ip_pools +from mtl_engine.media_files import yuv_files_422rfc10 + + +@pytest.mark.nightly +@pytest.mark.ptp +@pytest.mark.parametrize("application", ["gstreamer"]) +@pytest.mark.parametrize( + "media_file", + [ + yuv_files_422rfc10["Crosswalk_720p"], + yuv_files_422rfc10["ParkJoy_1080p"], + ], + indirect=["media_file"], + ids=["Crosswalk_720p", "ParkJoy_1080p"], +) +def test_st20p_ptp_gstreamer( + application, + app_factory, + hosts, + mtl_path, + setup_interfaces: InterfaceSetup, + test_config, + test_time, + pcap_capture, + media_file, +): + """Stream ST20P over GStreamer with onboard PTP enabled.""" + media_file_info, media_file_path = media_file + width = media_file_info["width"] + height = media_file_info["height"] + fps = media_file_info["fps"] + host = list(hosts.values())[0] + interfaces_list = setup_interfaces.get_interfaces_list_single( + test_config.get("interface_type", "VF") + ) + + # v210 pixel groups are six pixels wide; fall back to planar otherwise. + gst_format = "v210" if width % 6 == 0 else "I422_10LE" + media_dir = str(host.connection.path(media_file_path).parent) + input_file = media_create.create_video_file( + width=width, + height=height, + framerate=fps, + format=gst_format, + media_path=media_dir, + duration=3, + host=host, + ) + output_file = os.path.join(media_dir, "output_ptp_video.yuv") + + # PTP sync adds startup latency before the data window opens. + ptp_test_time = max(test_time, 30) + + app = app_factory(application) + app.create_command( + session_type="st20p", + nic_port_list=interfaces_list, + destination_ip=ip_pools.rx_multicast[0], + port=20000, + width=width, + height=height, + framerate=f"p{fps}", + gst_format=gst_format, + input_file=input_file, + output_file=output_file, + enable_ptp=True, + test_mode="multicast", + test_time=ptp_test_time, + ) + + try: + app.execute_test( + build=mtl_path, + test_time=ptp_test_time, + host=host, + netsniff=pcap_capture, + ) + finally: + media_create.remove_file(input_file, host=host) + media_create.remove_file(output_file, host=host) diff --git a/tests/validation/tests/single/st20p/test_resolutions.py b/tests/validation/tests/single/st20p/test_resolutions.py index 98421e0ff..6c8a752d5 100644 --- a/tests/validation/tests/single/st20p/test_resolutions.py +++ b/tests/validation/tests/single/st20p/test_resolutions.py @@ -1,13 +1,52 @@ # SPDX-License-Identifier: BSD-3-Clause # Copyright(c) 2026 Intel Corporation +import os + +import mtl_engine.media_creator as media_create import pytest from common.nicctl import InterfaceSetup from mtl_engine import ip_pools from mtl_engine.media_files import yuv_files_422rfc10 +from tests.xfail import SDBQ1971_conversion_v210_720p_error pytestmark = pytest.mark.verified +# GStreamer reads raw planar media (I422_10LE / v210), not the RFC4175-packed +# files RxTxApp/FFmpeg consume, so it self-provides a synthetic clip of the same +# resolution. Generating raw 4K/8K clips is impractical, so GStreamer only runs +# the <=1080p resolutions; RxTxApp/FFmpeg still sweep the full range. +_GST_MAX_HEIGHT = 1080 + + +def _prepare_gstreamer_media(host, info, media_file_path, request): + """Generate a synthetic planar clip for GStreamer; return (input, output, gst_format). + + The st20 plugin only emits v210 when the width is a multiple of six (pixel + groups are six pixels wide); otherwise it falls back to I422_10LE. The clip + is written next to the fixture-provided media file so it lands in the + managed media/ramdisk directory rather than a hardcoded path. + """ + width, height, fps = info["width"], info["height"], info["fps"] + gst_format = "v210" if width % 6 == 0 else "I422_10LE" + if gst_format == "v210": + SDBQ1971_conversion_v210_720p_error( + video_format=gst_format, resolution_height=height, request=request + ) + + media_dir = str(host.connection.path(media_file_path).parent) + input_file = media_create.create_video_file( + width=width, + height=height, + framerate=fps, + format=gst_format, + media_path=media_dir, + duration=3, + host=host, + ) + output_file = os.path.join(media_dir, "output_video.yuv") + return input_file, output_file, gst_format + @pytest.mark.nightly @pytest.mark.parametrize( @@ -15,6 +54,7 @@ [ "rxtxapp", "ffmpeg", + "gstreamer", ], ) @pytest.mark.parametrize( @@ -36,13 +76,15 @@ def test_st20p_resolutions( test_time, pcap_capture, media_file, + request, ): - """Test different video resolutions.""" + """Test different video resolutions across all framework adapters.""" media_file_info, media_file_path = media_file host = list(hosts.values())[0] interfaces_list = setup_interfaces.get_interfaces_list_single( test_config.get("interface_type", "VF") ) + height = media_file_info.get("height", 0) config_params = { "session_type": "st20p", @@ -50,30 +92,42 @@ def test_st20p_resolutions( "destination_ip": ip_pools.rx_multicast[0], "port": 20000, "width": media_file_info["width"], - "height": media_file_info["height"], + "height": height, "framerate": f"p{media_file_info['fps']}", "pixel_format": media_file_info["file_format"], "transport_format": media_file_info["format"], - "input_file": media_file_path, "test_mode": "multicast", "test_time": test_time, } - height = media_file_info.get("height", 0) - - if height >= 2160: - config_params.update( - {"pacing": "linear", "packing": "GPM_SL", "tx_no_chain": True} + gst_cleanup = [] + if application == "gstreamer": + if height > _GST_MAX_HEIGHT: + pytest.skip( + f"GStreamer st20p resolution sweep limited to <={_GST_MAX_HEIGHT}p " + "(raw planar clip generation)" + ) + input_file, output_file, gst_format = _prepare_gstreamer_media( + host, media_file_info, media_file_path, request ) - elif height >= 1080: - config_params.update({"pacing": "wide", "packing": "GPM", "tx_no_chain": False}) - else: config_params.update( - {"pacing": "narrow", "packing": "GPM", "tx_no_chain": False} + input_file=input_file, output_file=output_file, gst_format=gst_format ) - - app = app_factory(application) - app.create_command(**config_params) + gst_cleanup = [input_file, output_file] + else: + config_params["input_file"] = media_file_path + if height >= 2160: + config_params.update( + {"pacing": "linear", "packing": "GPM_SL", "tx_no_chain": True} + ) + elif height >= 1080: + config_params.update( + {"pacing": "wide", "packing": "GPM", "tx_no_chain": False} + ) + else: + config_params.update( + {"pacing": "narrow", "packing": "GPM", "tx_no_chain": False} + ) actual_test_time = test_time if height >= 2160: @@ -83,9 +137,16 @@ def test_st20p_resolutions( else: actual_test_time = max(test_time, 8) - app.execute_test( - build=mtl_path, - test_time=actual_test_time, - host=host, - netsniff=pcap_capture, - ) + app = app_factory(application) + app.create_command(**config_params) + + try: + app.execute_test( + build=mtl_path, + test_time=actual_test_time, + host=host, + netsniff=pcap_capture, + ) + finally: + for path in gst_cleanup: + media_create.remove_file(path, host=host) diff --git a/tests/validation/tests/single/st30p/test_st30p.py b/tests/validation/tests/single/st30p/test_st30p.py index 5f62f25d3..3bf5cb379 100644 --- a/tests/validation/tests/single/st30p/test_st30p.py +++ b/tests/validation/tests/single/st30p/test_st30p.py @@ -2,8 +2,10 @@ # Copyright(c) 2026 Intel Corporation import logging +import os from pathlib import Path +import mtl_engine.media_creator as media_create import pytest from common.integrity.integrity_runner import FileAudioIntegrityRunner from common.nicctl import InterfaceSetup @@ -457,3 +459,71 @@ def test_st30p_multicast( app.execute_test( build=mtl_path, test_time=test_time, host=host, netsniff=pcap_capture ) + + +@pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) +@pytest.mark.parametrize("audio_format", ["S8", "S16BE", "S24BE"]) +@pytest.mark.parametrize("audio_channel", [1, 2, 6, 8]) +@pytest.mark.parametrize("audio_rate", [44100, 48000, 96000]) +def test_st30p_gstreamer_audio_format( + application, + app_factory, + hosts, + mtl_path, + setup_interfaces: InterfaceSetup, + audio_format, + audio_channel, + audio_rate, + test_time, + test_config, + media_file, +): + """Sweep ST30 PCM formats, channel counts, and sampling rates over GStreamer. + + GStreamer drives the audio with audiotestsrc, so there is no reference clip + to compare against; this validates transport and caps negotiation only and + is therefore GStreamer-only (RxTxApp/FFmpeg integrity tests live above). + """ + if audio_rate == 96000 and ( + audio_channel == 8 + and (audio_format == "S16BE" or audio_format == "S24BE") + or audio_channel == 6 + and audio_format == "S24BE" + ): + pytest.skip(f"Audio {audio_format}/{audio_channel}ch invalid pkt_len; skipped") + + _media_file_info, media_file_path = media_file + if not media_file_path: + raise ValueError("ramdisk was not setup correctly for media_file fixture") + + host = list(hosts.values())[0] + input_file_path = os.path.join(media_file_path, "input_test_audio.pcm") + output_file_path = os.path.join(media_file_path, "output_test_audio.pcm") + interfaces_list = setup_interfaces.get_interfaces_list_single( + test_config.get("interface_type", "VF") + ) + + # Input path unused; audiotestsrc generates the signal internally. + app = app_factory(application) + app.create_command( + build=mtl_path, + session_type="st30", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, + audio_format=audio_format, + audio_channels=audio_channel, + audio_rate=audio_rate, + ) + + try: + app.execute_test( + build=mtl_path, + test_time=test_time, + host=host, + tx_first=False, + sleep_interval=1, + ) + finally: + media_create.remove_file(output_file_path, host=host) diff --git a/tests/validation/tests/single/gstreamer/anc_format/test_anc_format.py b/tests/validation/tests/single/st40p/test_st40p_anc.py similarity index 79% rename from tests/validation/tests/single/gstreamer/anc_format/test_anc_format.py rename to tests/validation/tests/single/st40p/test_st40p_anc.py index b1aaceccc..cd7f99fc1 100755 --- a/tests/validation/tests/single/gstreamer/anc_format/test_anc_format.py +++ b/tests/validation/tests/single/st40p/test_st40p_anc.py @@ -35,69 +35,6 @@ def _frame_info_path(base_dir=None): ) -def _append_redundant_params( - pipeline: list[str], - dev_port_red: str, - dev_ip_red: str, - ip_red: str, - udp_port_red: int, -): - primary_udp = _find_param_value(pipeline, "udp-port") - primary_udp_int = None - try: - primary_udp_int = int(primary_udp) if primary_udp is not None else None - except (TypeError, ValueError): - primary_udp_int = None - - GstreamerApp.add_redundant_params( - pipeline, - dev_port_red=dev_port_red, - dev_ip_red=dev_ip_red, - ip_red=ip_red, - udp_port_red=udp_port_red, - port_red=dev_port_red, - primary_udp_port=primary_udp_int, - ) - - -def _find_param_value(pipeline: list[str], key: str) -> str | None: - prefix = f"{key}=" - for item in pipeline: - if item.startswith(prefix): - return item[len(prefix) :] - return None - - -def _find_param_indices(pipeline: list[str], keys: list[str]) -> dict[str, int]: - indices: dict[str, int] = {} - for idx, item in enumerate(pipeline): - for key in keys: - if item.startswith(f"{key}="): - indices[key] = idx - return indices - - -def _log_redundant_debug( - test_name: str, - pipeline: list[str], - tx_ports: list[str], - rx_ports: list[str], -): - element_idx = None - for idx, item in enumerate(pipeline): - if item in ("mtl_st40p_tx", "mtl_st40p_rx"): - element_idx = idx - break - keys = ["dev-port-red", "port-red", "dev-ip-red", "ip-red", "udp-port-red"] - indices = _find_param_indices(pipeline, keys) - values = {key: _find_param_value(pipeline, key) for key in keys} - log_info( - f"[{test_name}] Redundant debug: element_idx={element_idx}, indices={indices}, " - f"values={values}, tx_ports={tx_ports}, rx_ports={rx_ports}, " - f"ip_pools_tx={ip_pools.tx}, ip_pools_rx={ip_pools.rx}" - ) - - def _log_vf_link_state(test_name: str, host, bdfs: list[str]) -> None: for bdf in bdfs: bind = run(f"dpdk-devbind.py -s | grep -F '{bdf}'", host=host) @@ -228,10 +165,13 @@ def setup_paths(media_file): @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [24, 25, 30, 50, 60, 100, 120]) @pytest.mark.parametrize("file_size_kb", [10, 100]) @pytest.mark.parametrize("framebuff", [3]) def test_st40p_fps_size( + application, + app_factory, hosts, mtl_path, media, @@ -285,26 +225,23 @@ def test_st40p_fps_size( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, tx_fps=fps, tx_did=67, tx_sdid=2, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=15, + rx_timeout=15, ) expectation = ( @@ -314,16 +251,13 @@ def test_st40p_fps_size( try: with _test_summary("test_st40p_fps_size", expectation): - GstreamerApp.execute_test( + app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) finally: @@ -333,7 +267,10 @@ def test_st40p_fps_size( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_redundant_progressive( + application, + app_factory, hosts, mtl_path, media, @@ -368,29 +305,25 @@ def test_st40p_redundant_progressive( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, tx_fps=24, tx_did=67, tx_sdid=2, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, ) _log_vf_link_state( "test_st40p_redundant_progressive_link", @@ -403,44 +336,17 @@ def test_st40p_redundant_progressive( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = "Redundant progressive ST40P logs pkts_recv_p/r with deduped totals" try: with _test_summary("test_st40p_redundant_progressive", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -454,7 +360,10 @@ def test_st40p_redundant_progressive( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_redundant_progressive_gap( + application, + app_factory, hosts, mtl_path, media, @@ -488,10 +397,14 @@ def test_st40p_redundant_progressive_gap( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -500,19 +413,11 @@ def test_st40p_redundant_progressive_gap( tx_sdid=2, tx_test_mode="seq-gap", tx_test_pkt_count=200, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_gap_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, ) _log_vf_link_state( "test_st40p_redundant_progressive_gap_link", @@ -525,46 +430,19 @@ def test_st40p_redundant_progressive_gap( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_gap_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = ( "Redundant progressive ST40P with seq-gap injects and logs pkts_recv_p/r" ) try: with _test_summary("test_st40p_redundant_progressive_gap", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -578,7 +456,10 @@ def test_st40p_redundant_progressive_gap( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_redundant_progressive_split( + application, + app_factory, hosts, mtl_path, media, @@ -613,10 +494,14 @@ def test_st40p_redundant_progressive_split( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -624,19 +509,11 @@ def test_st40p_redundant_progressive_split( tx_did=67, tx_sdid=2, tx_split_anc_by_pkt=True, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_split_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, ) _log_vf_link_state( "test_st40p_redundant_progressive_split_link", @@ -649,44 +526,17 @@ def test_st40p_redundant_progressive_split( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_split_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = "Redundant split-mode ST40P logs pkts_recv_p/r with deduped totals" try: with _test_summary("test_st40p_redundant_progressive_split", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -700,7 +550,10 @@ def test_st40p_redundant_progressive_split( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_redundant_progressive_split_gap( + application, + app_factory, hosts, mtl_path, media, @@ -734,10 +587,14 @@ def test_st40p_redundant_progressive_split_gap( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -747,19 +604,11 @@ def test_st40p_redundant_progressive_split_gap( tx_split_anc_by_pkt=True, tx_test_mode="seq-gap", tx_test_pkt_count=200, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_split_gap_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, ) _log_vf_link_state( "test_st40p_redundant_progressive_split_gap_link", @@ -772,44 +621,17 @@ def test_st40p_redundant_progressive_split_gap( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40p_redundant_progressive_split_gap_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = "Redundant split ST40P with seq-gap injects and logs pkts_recv_p/r" try: with _test_summary("test_st40p_redundant_progressive_split_gap", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -823,7 +645,10 @@ def test_st40p_redundant_progressive_split_gap( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40i_redundant_split( + application, + app_factory, hosts, mtl_path, media, @@ -858,10 +683,14 @@ def test_st40i_redundant_split( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -870,19 +699,12 @@ def test_st40i_redundant_split( tx_sdid=2, tx_split_anc_by_pkt=True, tx_interlaced=True, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40i_redundant_split_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, + rx_interlaced=True, ) _log_vf_link_state( "test_st40i_redundant_split_link", @@ -895,47 +717,19 @@ def test_st40i_redundant_split( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - rx_interlaced=True, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40i_redundant_split_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = ( "Redundant interlaced split-mode ST40 logs pkts_recv_p/r with deduped totals" ) try: with _test_summary("test_st40i_redundant_split", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -949,7 +743,10 @@ def test_st40i_redundant_split( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40i_redundant_split_gap( + application, + app_factory, hosts, mtl_path, media, @@ -983,10 +780,14 @@ def test_st40i_redundant_split_gap( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=tx_ports[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=tx_ports + rx_ports, + input_file=input_file_path, + output_file=output_file_path, + redundant=True, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -997,19 +798,12 @@ def test_st40i_redundant_split_gap( tx_interlaced=True, tx_test_mode="seq-gap", tx_test_pkt_count=200, - ) - _append_redundant_params( - tx_config, - dev_port_red=tx_ports[1], - dev_ip_red=ip_pools.rx[1], - ip_red=ip_pools.tx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40i_redundant_split_gap_tx", - tx_config, - tx_ports, - rx_ports, + rx_payload_type=113, + rx_queues=4, + rx_framebuff_cnt=3, + rx_timeout=20, + frame_info_path=frame_info_path, + rx_interlaced=True, ) _log_vf_link_state( "test_st40i_redundant_split_gap_link", @@ -1022,47 +816,19 @@ def test_st40i_redundant_split_gap( tx_ports + rx_ports, ) - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=rx_ports[0], - output_path=output_file_path, - rx_payload_type=113, - rx_queues=4, - rx_framebuff_cnt=3, - timeout=20, - frame_info_path=frame_info_path, - rx_interlaced=True, - ) - _append_redundant_params( - rx_config, - dev_port_red=rx_ports[1], - dev_ip_red=ip_pools.tx[1], - ip_red=ip_pools.rx[1], - udp_port_red=40001, - ) - _log_redundant_debug( - "test_st40i_redundant_split_gap_rx", - rx_config, - tx_ports, - rx_ports, - ) - expectation = ( "Redundant interlaced split ST40 with seq-gap injects and logs pkts_recv_p/r" ) try: with _test_summary("test_st40i_redundant_split_gap", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 12), host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) info_dump = run(f"cat {frame_info_path}", host=host) @@ -1076,10 +842,13 @@ def test_st40i_redundant_split_gap( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [60]) @pytest.mark.parametrize("file_size_kb", [100]) @pytest.mark.parametrize("framebuff", [1, 3, 6, 12]) def test_st40p_framebuff( + application, + app_factory, hosts, mtl_path, media, @@ -1133,26 +902,23 @@ def test_st40p_framebuff( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, tx_fps=fps, tx_did=67, tx_sdid=2, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=timeout_period + 10, + rx_timeout=timeout_period + 10, ) expectation = ( @@ -1162,16 +928,13 @@ def test_st40p_framebuff( try: with _test_summary("test_st40p_framebuff", expectation): - GstreamerApp.execute_test( + app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=timeout_period, + fail_on_error=False, log_frame_info=True, ) finally: @@ -1189,9 +952,12 @@ def test_st40p_framebuff( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [24, 25, 30, 50, 60, 100, 120]) @pytest.mark.parametrize("framebuff", [1, 3, 6, 12]) def test_st40p_format_8331( + application, + app_factory, hosts, mtl_path, media, @@ -1239,10 +1005,13 @@ def test_st40p_format_8331( input_file_path, output_file_path = setup_paths(media_file) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, @@ -1250,16 +1019,10 @@ def test_st40p_format_8331( tx_did=67, tx_sdid=2, tx_rfc8331=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=timeout_period + 10, + rx_timeout=timeout_period + 10, capture_metadata=True, ) @@ -1270,16 +1033,13 @@ def test_st40p_format_8331( try: with _test_summary("test_st40p_format_8331", expectation): - GstreamerApp.execute_test( + app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=timeout_period, + fail_on_error=False, log_frame_info=True, ) finally: @@ -1289,10 +1049,13 @@ def test_st40p_format_8331( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [25, 50, 60]) @pytest.mark.parametrize("file_size_kb", [10, 100]) @pytest.mark.parametrize("framebuff", [3, 6]) def test_st40i_basic( + application, + app_factory, hosts, mtl_path, media, @@ -1343,10 +1106,13 @@ def test_st40i_basic( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, @@ -1354,16 +1120,10 @@ def test_st40i_basic( tx_did=67, tx_sdid=2, tx_interlaced=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=15, + rx_timeout=15, rx_interlaced=True, ) @@ -1374,16 +1134,13 @@ def test_st40i_basic( try: with _test_summary("test_st40i_basic", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) finally: @@ -1392,9 +1149,12 @@ def test_st40i_basic( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [50]) @pytest.mark.parametrize("framebuff", [3]) def test_st40i_rfc8331( + application, + app_factory, hosts, mtl_path, media, @@ -1443,10 +1203,13 @@ def test_st40i_rfc8331( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, @@ -1455,16 +1218,10 @@ def test_st40i_rfc8331( tx_sdid=2, tx_rfc8331=True, tx_interlaced=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=20, + rx_timeout=20, capture_metadata=True, rx_interlaced=True, ) @@ -1476,16 +1233,13 @@ def test_st40i_rfc8331( try: with _test_summary("test_st40i_rfc8331", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) finally: @@ -1494,10 +1248,13 @@ def test_st40i_rfc8331( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) @pytest.mark.parametrize("fps", [50]) @pytest.mark.parametrize("file_size_kb", [10]) @pytest.mark.parametrize("framebuff", [3]) def test_st40i_interlace_flag_mismatch( + application, + app_factory, hosts, mtl_path, media, @@ -1546,10 +1303,13 @@ def test_st40i_interlace_flag_mismatch( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=framebuff, @@ -1557,16 +1317,10 @@ def test_st40i_interlace_flag_mismatch( tx_did=67, tx_sdid=2, tx_interlaced=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=framebuff, - timeout=15, + rx_timeout=15, rx_interlaced=False, ) @@ -1577,16 +1331,13 @@ def test_st40i_interlace_flag_mismatch( try: with _test_summary("test_st40i_interlace_flag_mismatch", expectation): - result = GstreamerApp.execute_test( + result = app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) assert result, "Interlace mismatch unexpectedly failed" @@ -1596,7 +1347,10 @@ def test_st40i_interlace_flag_mismatch( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_interlace_auto_detect_reset( + application, + app_factory, hosts, mtl_path, media, @@ -1628,10 +1382,15 @@ def test_st40p_interlace_auto_detect_reset( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + frame_info_path = _frame_info_path(os.path.dirname(output_file_path)) + + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -1639,17 +1398,9 @@ def test_st40p_interlace_auto_detect_reset( tx_did=67, tx_sdid=2, tx_interlaced=True, - ) - - frame_info_path = _frame_info_path(os.path.dirname(output_file_path)) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, - timeout=15, + rx_timeout=15, rx_framebuff_cnt=3, rx_interlaced=False, frame_info_path=frame_info_path, @@ -1659,25 +1410,24 @@ def test_st40p_interlace_auto_detect_reset( try: with _test_summary("test_st40p_interlace_auto_detect_reset", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) # Force a sequence gap to reset auto-detect state and prove cadence re-learns gap_frame_info_path = _frame_info_path(os.path.dirname(output_file_path)) - tx_config_gap = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -1688,15 +1438,9 @@ def test_st40p_interlace_auto_detect_reset( tx_split_anc_by_pkt=True, tx_test_mode="seq-gap", tx_test_pkt_count=200, - ) - - rx_config_gap = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, - timeout=15, + rx_timeout=15, rx_framebuff_cnt=3, rx_interlaced=False, frame_info_path=gap_frame_info_path, @@ -1709,16 +1453,13 @@ def test_st40p_interlace_auto_detect_reset( with _test_summary( "test_st40p_interlace_auto_detect_reset_gap", reset_expectation ): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config_gap, - rx_command=rx_config_gap, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, log_frame_info=True, ) @@ -1739,7 +1480,10 @@ def test_st40p_interlace_auto_detect_reset( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_rx_timeout( + application, + app_factory, hosts, mtl_path, media, @@ -1781,42 +1525,36 @@ def test_st40p_rx_timeout( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, tx_fps=60, tx_did=67, tx_sdid=2, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=1, # force receiver timeout + rx_timeout=1, # force receiver timeout ) expectation = "RX timeout triggers failure with no payload capture" try: with _test_summary("test_st40p_rx_timeout", expectation): - result = GstreamerApp.execute_test( + result = app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=2, host=host, tx_first=False, sleep_interval=1, + fail_on_error=False, suppress_fail_logs=True, log_frame_info=True, ) @@ -1827,7 +1565,10 @@ def test_st40p_rx_timeout( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_split_mode_frame_info_logging( + application, + app_factory, hosts, mtl_path, media, @@ -1873,10 +1614,13 @@ def test_st40p_split_mode_frame_info_logging( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -1884,16 +1628,10 @@ def test_st40p_split_mode_frame_info_logging( tx_did=67, tx_sdid=2, tx_split_anc_by_pkt=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=15, + rx_timeout=15, frame_info_path=frame_info_path, rx_rtp_ring_size=2048, ) @@ -1902,16 +1640,13 @@ def test_st40p_split_mode_frame_info_logging( try: with _test_summary("test_st40p_split_mode_frame_info_logging", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=6, host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) @@ -1927,7 +1662,10 @@ def test_st40p_split_mode_frame_info_logging( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_split_mode_invalid_rtp_ring_rejected( + application, + app_factory, hosts, mtl_path, media, @@ -1972,14 +1710,19 @@ def test_st40p_split_mode_invalid_rtp_ring_rejected( ) # Non power-of-two ring size should be rejected by plugin - rx_cmd = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, + tx_payload_type=113, + tx_queues=4, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=15, + rx_timeout=15, frame_info_path=frame_info_path, rx_rtp_ring_size=100, ) @@ -1990,7 +1733,7 @@ def test_st40p_split_mode_invalid_rtp_ring_rejected( with _test_summary( "test_st40p_split_mode_invalid_rtp_ring_rejected", expectation ): - rx_proc = run(" ".join(rx_cmd), cwd=mtl_path, timeout=10, host=host) + rx_proc = run(app.command, cwd=mtl_path, timeout=10, host=host) assert rx_proc.return_code != 0 finally: media_create.remove_file(input_file_path, host=host) @@ -1999,7 +1742,10 @@ def test_st40p_split_mode_invalid_rtp_ring_rejected( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_split_mode_pacing_respected( + application, + app_factory, hosts, mtl_path, media, @@ -2049,10 +1795,13 @@ def test_st40p_split_mode_pacing_respected( tx_fps = 60 - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2063,31 +1812,22 @@ def test_st40p_split_mode_pacing_respected( tx_test_mode="paced", tx_test_pkt_count=8, tx_test_pacing_ns=200000, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=15, + rx_timeout=15, frame_info_path=frame_info_path, ) try: with _test_summary("test_st40p_split_mode_pacing_respected", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=3, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2126,7 +1866,10 @@ def test_st40p_split_mode_pacing_respected( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_rx_missing_marker_no_ready( + application, + app_factory, hosts, mtl_path, media, @@ -2173,10 +1916,13 @@ def test_st40p_rx_missing_marker_no_ready( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2186,16 +1932,10 @@ def test_st40p_rx_missing_marker_no_ready( tx_split_anc_by_pkt=True, tx_test_mode="no-marker", tx_test_pkt_count=1, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=20, + rx_timeout=20, frame_info_path=frame_info_path, ) @@ -2203,16 +1943,13 @@ def test_st40p_rx_missing_marker_no_ready( try: with _test_summary("test_st40p_rx_missing_marker_no_ready", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=3, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2226,7 +1963,10 @@ def test_st40p_rx_missing_marker_no_ready( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_rx_seq_loss_logged( + application, + app_factory, hosts, mtl_path, media, @@ -2273,10 +2013,13 @@ def test_st40p_rx_seq_loss_logged( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2286,16 +2029,10 @@ def test_st40p_rx_seq_loss_logged( tx_split_anc_by_pkt=True, tx_test_mode="seq-gap", tx_test_pkt_count=2, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=20, + rx_timeout=20, frame_info_path=frame_info_path, ) @@ -2303,16 +2040,13 @@ def test_st40p_rx_seq_loss_logged( try: with _test_summary("test_st40p_rx_seq_loss_logged", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=3, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2354,7 +2088,10 @@ def test_st40p_rx_seq_loss_logged( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_rx_bad_parity_drops_payload( + application, + app_factory, hosts, mtl_path, media, @@ -2399,10 +2136,13 @@ def test_st40p_rx_bad_parity_drops_payload( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2412,16 +2152,10 @@ def test_st40p_rx_bad_parity_drops_payload( tx_split_anc_by_pkt=True, tx_test_mode="bad-parity", tx_test_pkt_count=1, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=20, + rx_timeout=20, frame_info_path=frame_info_path, ) @@ -2429,16 +2163,13 @@ def test_st40p_rx_bad_parity_drops_payload( try: with _test_summary("test_st40p_rx_bad_parity_drops_payload", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=3, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2463,7 +2194,10 @@ def test_st40p_rx_bad_parity_drops_payload( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_rx_multi_packet_field_accumulates( + application, + app_factory, hosts, mtl_path, media, @@ -2510,10 +2244,13 @@ def test_st40p_rx_multi_packet_field_accumulates( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2523,16 +2260,10 @@ def test_st40p_rx_multi_packet_field_accumulates( tx_split_anc_by_pkt=True, tx_test_mode="paced", tx_test_pkt_count=3, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=40, + rx_timeout=40, frame_info_path=frame_info_path, ) @@ -2540,16 +2271,13 @@ def test_st40p_rx_multi_packet_field_accumulates( try: with _test_summary("test_st40p_rx_multi_packet_field_accumulates", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=5, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2567,7 +2295,10 @@ def test_st40p_rx_multi_packet_field_accumulates( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_split_padding_alignment_boundary( + application, + app_factory, hosts, mtl_path, media, @@ -2603,10 +2334,13 @@ def test_st40p_split_padding_alignment_boundary( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2617,16 +2351,10 @@ def test_st40p_split_padding_alignment_boundary( tx_test_mode="paced", tx_test_pkt_count=2, tx_test_pacing_ns=200000, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=20, + rx_timeout=20, frame_info_path=frame_info_path, ) @@ -2636,16 +2364,13 @@ def test_st40p_split_padding_alignment_boundary( try: with _test_summary("test_st40p_split_padding_alignment_boundary", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=test_time, host=host, tx_first=False, sleep_interval=3, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2664,7 +2389,10 @@ def test_st40p_split_padding_alignment_boundary( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40p_split_tx_mtu_guard_no_stall( + application, + app_factory, hosts, mtl_path, media, @@ -2701,10 +2429,13 @@ def test_st40p_split_tx_mtu_guard_no_stall( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=2, @@ -2714,16 +2445,10 @@ def test_st40p_split_tx_mtu_guard_no_stall( tx_split_anc_by_pkt=True, tx_test_mode="paced", tx_test_pkt_count=8, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=2, - timeout=10, + rx_timeout=10, frame_info_path=frame_info_path, ) @@ -2731,16 +2456,13 @@ def test_st40p_split_tx_mtu_guard_no_stall( try: with _test_summary("test_st40p_split_tx_mtu_guard_no_stall", expectation): - result = GstreamerApp.execute_test( + result = app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=min(test_time, 6), host=host, tx_first=False, sleep_interval=2, + fail_on_error=False, skip_file_compare=True, log_frame_info=True, ) @@ -2753,7 +2475,10 @@ def test_st40p_split_tx_mtu_guard_no_stall( @pytest.mark.nightly +@pytest.mark.parametrize("application", ["gstreamer"]) def test_st40i_split_mode_frame_info_logging( + application, + app_factory, hosts, mtl_path, media, @@ -2799,10 +2524,13 @@ def test_st40i_split_mode_frame_info_logging( host=host, ) - tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline( + app = app_factory(application) + app.create_command( build=mtl_path, - nic_port_list=interfaces_list[0], - input_path=input_file_path, + session_type="st40p", + nic_port_list=interfaces_list, + input_file=input_file_path, + output_file=output_file_path, tx_payload_type=113, tx_queues=4, tx_framebuff_cnt=3, @@ -2811,16 +2539,10 @@ def test_st40i_split_mode_frame_info_logging( tx_sdid=2, tx_split_anc_by_pkt=True, tx_interlaced=True, - ) - - rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline( - build=mtl_path, - nic_port_list=interfaces_list[1], - output_path=output_file_path, rx_payload_type=113, rx_queues=4, rx_framebuff_cnt=3, - timeout=15, + rx_timeout=15, frame_info_path=frame_info_path, rx_interlaced=True, ) @@ -2829,16 +2551,13 @@ def test_st40i_split_mode_frame_info_logging( try: with _test_summary("test_st40i_split_mode_frame_info_logging", expectation): - assert GstreamerApp.execute_test( + assert app.execute_test( build=mtl_path, - tx_command=tx_config, - rx_command=rx_config, - input_file=input_file_path, - output_file=output_file_path, test_time=6, host=host, tx_first=False, sleep_interval=4, + fail_on_error=False, log_frame_info=True, ) diff --git a/tests/validation/tests/validation/test_gstreamer_adapter.py b/tests/validation/tests/validation/test_gstreamer_adapter.py new file mode 100644 index 000000000..149be6833 --- /dev/null +++ b/tests/validation/tests/validation/test_gstreamer_adapter.py @@ -0,0 +1,149 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2026 Intel Corporation + +"""Characterization tests for the GStreamer adapter's parameter translation. + +No hardware or host is touched. Each case asserts the UNIVERSAL_PARAMS +vocabulary (shared with RxTxApp/FFmpeg) and the ``enable_ptp`` flag translate +onto exactly the token lists produced by the procedural ``GstreamerApp`` +builders (the single source of truth), so a common parametrized test can drive +GStreamer identically to the other frameworks. +""" + +import pytest +from mtl_engine import GstreamerApp, ip_pools +from mtl_engine.gstreamer import GStreamer + +BUILD = "/fake/build" + + +@pytest.fixture(autouse=True) +def _ip_pools(): + for pool in ( + ip_pools.rx, + ip_pools.rx_multicast, + ip_pools.tx, + ip_pools.tx_r, + ip_pools.rx_r, + ): + pool.clear() + ip_pools.init(session_id=1) + yield + + +def test_st20p_universal_params_translate_to_builders(): + """UNIVERSAL_PARAMS vocabulary (transport_format + ``pNN`` framerate) maps + onto the same tokens the GStreamer-only vocabulary produced.""" + nic = ["0000:01:00.0", "0000:01:00.1"] + app = GStreamer(BUILD) + app.create_command( + session_type="st20p", + nic_port_list=nic, + width=1920, + height=1080, + framerate="p25", + transport_format="YUV_422_10bit", + pixel_format="YUV422PLANAR10LE", + input_file="/in.yuv", + output_file="/out.yuv", + test_mode="multicast", + ) + + gst_format = GstreamerApp.video_format_change("YUV_422_10bit") + tx_golden = GstreamerApp.setup_gstreamer_st20p_tx_pipeline( + build=BUILD, + nic_port_list=nic[0], + input_path="/in.yuv", + width=1920, + height=1080, + framerate="25", + format=gst_format, + tx_payload_type=112, + tx_queues=4, + ) + rx_golden = GstreamerApp.setup_gstreamer_st20p_rx_pipeline( + build=BUILD, + nic_port_list=nic[1], + output_path="/out.yuv", + width=1920, + height=1080, + framerate="25", + format=gst_format, + rx_payload_type=112, + rx_queues=4, + ) + assert app._tx_command == tx_golden + assert app._rx_command == rx_golden + + +def test_st20p_enable_ptp_emits_token(): + """``enable_ptp=True`` appends ``enable-ptp=true`` to both TX and RX; the + default keeps it absent.""" + nic = ["0000:01:00.0", "0000:01:00.1"] + app = GStreamer(BUILD) + app.create_command( + session_type="st20p", + nic_port_list=nic, + width=1920, + height=1080, + framerate="p25", + gst_format="I422_10LE", + input_file="/in.yuv", + output_file="/out.yuv", + enable_ptp=True, + ) + assert "enable-ptp=true" in app._tx_command + assert "enable-ptp=true" in app._rx_command + + app_default = GStreamer(BUILD) + app_default.create_command( + session_type="st20p", + nic_port_list=nic, + width=1920, + height=1080, + framerate="p25", + gst_format="I422_10LE", + input_file="/in.yuv", + output_file="/out.yuv", + ) + assert "enable-ptp=true" not in app_default._tx_command + assert "enable-ptp=true" not in app_default._rx_command + + +def test_st30p_universal_audio_params_translate_to_builders(): + """Universal audio vocabulary (``PCM24`` / ``["U02"]`` / ``48kHz`` and the + ``st30p`` alias) maps onto the S24BE / 2ch / 48000 GStreamer builders.""" + nic = ["0000:01:00.0", "0000:01:00.1"] + app = GStreamer(BUILD) + app.create_command( + session_type="st30p", + nic_port_list=nic, + input_file="/in.pcm", + output_file="/out.pcm", + audio_format="PCM24", + audio_channels=["U02"], + audio_sampling="48kHz", + ) + + tx_golden = GstreamerApp.setup_gstreamer_st30_tx_pipeline( + build=BUILD, + nic_port_list=nic[0], + input_path="/in.pcm", + tx_payload_type=111, + tx_queues=4, + audio_format="S24BE", + channels=2, + sampling=48000, + ) + rx_golden = GstreamerApp.setup_gstreamer_st30_rx_pipeline( + build=BUILD, + nic_port_list=nic[1], + output_path="/out.pcm", + rx_payload_type=111, + rx_queues=4, + rx_audio_format=GstreamerApp.audio_format_change("S24BE", rx_side=True), + rx_channels=2, + rx_sampling=48000, + ) + assert app._tx_command == tx_golden + assert app._rx_command == rx_golden