Fix: tests: stabilize gst video/audio validation suites (#1347)

- teach GstreamerApp to handle I422_10LE properly: explicit caps,
per-format blocksize calculation, and RX pixel-format remapping so
rawvideoparse gaps stop breaking UHD fixtures
- normalize audio-cap names on both TX and RX so the st30 pipeline
request the PCM width the plugin expects
- guard the single-host tests against missing SR-IOV VFs and split TX/RX
VF usage to avoid NIC contention
- update the single- and dual-host video resolution suites to clone
media descriptors, only request v210 when the width is divisible by six,
and keep the 720p xfail scoped to its true height-based condition

Author: moleksy

tests/validation/mtl_engine/GstreamerApp.py

@@ -5,6 +5,7 @@
 import os
 import re
 import time
+from typing import Optional
 
 from mtl_engine import ip_pools
 
@@ -68,6 +69,16 @@ def setup_gstreamer_plugins_paths(build):
     return ":".join(plugin_paths)
 
 
+def calculate_frame_size(pixel_format: str, width: int, height: int) -> Optional[int]:
+    """Return the byte size of a single frame for formats that lack parser support."""
+
+    format_frame_size = {
+        "I422_10LE": width * height * 4,
+    }
+
+    return format_frame_size.get(pixel_format)
+
+
 def fract_format(framerate: str) -> str:
     """
     Convert framerate to proper fractional format for GStreamer.
@@ -139,8 +150,22 @@ def setup_gstreamer_st20p_tx_pipeline(
             ]
         )
     elif format == "I422_10LE":
+        # Apply explicit caps instead of rawvideoparse, which lacks this format
+        caps = (
+            f"video/x-raw,format={format},height={height},"
+            f"width={width},framerate={framerate}"
+        )
+        frame_size = calculate_frame_size(format, width, height)
+        filesrc_args = ["filesrc", f"location={input_path}"]
+        if frame_size:
+            filesrc_args.append(f"blocksize={frame_size}")
         pipeline_command.extend(
-            ["filesrc", f"location={input_path}", f"blocksize={width * height * 10}"]
+            filesrc_args
+            + [
+                "!",
+                caps,
+                "!",
+            ]
         )
 
     pipeline_command.extend(["mtl_st20p_tx", f"tx-queues={tx_queues}"])
@@ -182,14 +207,15 @@ def setup_gstreamer_st20p_rx_pipeline(
     )
 
     framerate = fract_format(framerate)
+    mtl_pixel_format = map_gstreamer_to_mtl_pixel_format(format)
 
     # st20 rx GStreamer command line
     pipeline_command = [
         "gst-launch-1.0",
         "-v",
         "mtl_st20p_rx",
         f"rx-queues={rx_queues}",
-        f"rx-pixel-format={format}",
+        f"rx-pixel-format={mtl_pixel_format}",
         f"rx-height={height}",
         f"rx-width={width}",
         f"rx-fps={framerate}",
@@ -720,21 +746,41 @@ def video_format_change(file_format):
         return file_format
 
 
+def map_gstreamer_to_mtl_pixel_format(file_format: str) -> str:
+    """Translate GStreamer caps names into the strings expected by the MTL plugin."""
+
+    format_map = {
+        "I422_10LE": "YUV422PLANAR10LE",
+    }
+
+    return format_map.get(file_format, file_format)
+
+
 def audio_format_change(file_format, rx_side: bool = False):
+    """Translate GST caps strings to the plugin naming conventions."""
+
+    fmt = (file_format or "").lower()
+
     if rx_side:
-        if file_format == "s8":
-            return "PCM8"
-        elif file_format == "s16le":
-            return "PCM16"
-        else:
-            return "PCM24"
-    else:
-        if file_format == "s8":
-            return 8
-        elif file_format == "s16le":
-            return 16
-        else:
-            return 24
+        rx_map = {
+            "s8": "PCM8",
+            "u8": "PCM8",
+            "s16le": "PCM16",
+            "s16be": "PCM16",
+            "s24le": "PCM24",
+            "s24be": "PCM24",
+        }
+        return rx_map.get(fmt, "PCM24")
+
+    tx_map = {
+        "s8": 8,
+        "u8": 8,
+        "s16le": 16,
+        "s16be": 16,
+        "s24le": 24,
+        "s24be": 24,
+    }
+    return tx_map.get(fmt, 24)
 
 
 def get_case_id() -> str:

tests/validation/tests/dual/gstreamer/video_resolution/test_video_resolution_dual.py

@@ -24,8 +24,13 @@ def test_video_resolutions_dual(
     prepare_ramdisk,
 ):
     """Test GStreamer ST20P video resolution in dual host configuration."""
-    video_file = yuv_files[file]
-    video_file["format"] = "v210"
+    video_file = yuv_files[file].copy()
+
+    gst_format = (
+        "v210"
+        if video_file["width"] % 6 == 0
+        else GstreamerApp.video_format_change(video_file["format"])
+    )
 
     # Get TX and RX hosts
     host_list = list(hosts.values())
@@ -35,18 +40,19 @@ def test_video_resolutions_dual(
     tx_host = host_list[0]
     rx_host = host_list[1]
 
-    SDBQ1971_conversion_v210_720p_error(
-        video_format=video_file["format"],
-        resolution_width=video_file["height"],
-        request=request,
-    )
+    if gst_format == "v210":
+        SDBQ1971_conversion_v210_720p_error(
+            video_format=gst_format,
+            resolution_height=video_file["height"],
+            request=request,
+        )
 
     # Create input file on TX host
     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"]),
+        format=gst_format,
         media_path=media,
         duration=2,
         host=tx_host,
@@ -63,7 +69,7 @@ def test_video_resolutions_dual(
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=GstreamerApp.video_format_change(video_file["format"]),
+        format=gst_format,
         tx_payload_type=112,
         tx_queues=4,
     )
@@ -76,7 +82,7 @@ def test_video_resolutions_dual(
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=GstreamerApp.video_format_change(video_file["format"]),
+        format=gst_format,
         rx_payload_type=112,
         rx_queues=4,
     )

tests/validation/tests/single/gstreamer/video_resolution/test_video_resolution.py

@@ -24,26 +24,35 @@ def test_video_resolutions(
     test_config,
     prepare_ramdisk,
 ):
-    video_file = yuv_files[file]
-    video_file["format"] = "v210"
+    video_file = yuv_files[file].copy()
+
+    # 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")
     )
 
-    SDBQ1971_conversion_v210_720p_error(
-        video_format=video_file["format"],
-        resolution_width=video_file["height"],
-        request=request,
-    )
+    if gst_format == "v210":
+        SDBQ1971_conversion_v210_720p_error(
+            video_format=gst_format,
+            resolution_height=video_file["height"],
+            request=request,
+        )
 
     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"]),
+        format=gst_format,
         media_path=media,
         duration=3,
         host=host,
@@ -56,7 +65,7 @@ def test_video_resolutions(
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=GstreamerApp.video_format_change(video_file["format"]),
+        format=gst_format,
         tx_payload_type=112,
         tx_queues=4,
     )
@@ -68,7 +77,7 @@ def test_video_resolutions(
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=GstreamerApp.video_format_change(video_file["format"]),
+        format=gst_format,
         rx_payload_type=112,
         rx_queues=4,
     )

tests/validation/tests/xfail.py

@@ -29,9 +29,9 @@ def SDBQ1002_pg_format_error_check(video_format: str, pg_format: str, request):
 
 
 def SDBQ1971_conversion_v210_720p_error(
-    video_format: str, resolution_width: int, request
+    video_format: str, resolution_height: int, request
 ):
-    if video_format == "v210" and resolution_width == 720:
+    if video_format == "v210" and resolution_height == 720:
         add_issue(
             "XFAIL: SDBQ-1971 - Conversion from v210 format does not work on 720p",
             request,