Add dual host GStreamer tests

Author: Falron98

tests/validation/mtl_engine/GstreamerApp.py

@@ -0,0 +1,70 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024-2025 Intel Corporation
+
+import os
+
+import mtl_engine.media_creator as media_create
+import pytest
+from mtl_engine import GstreamerApp
+
+
+@pytest.mark.parametrize("fps", [24, 25, 30, 50, 60])
+@pytest.mark.parametrize("file_size_kb", [10, 100])
+@pytest.mark.parametrize("framebuff", [3])
+def test_st40p_fps_size_dual(
+    hosts,
+    build,
+    media,
+    nic_port_list,
+    file_size_kb,
+    fps,
+    framebuff,
+    test_time,
+    test_config,
+    prepare_ramdisk,
+):
+    """Test GStreamer ST40P ANC format in dual host configuration."""
+    # Get TX and RX hosts
+    host_list = list(hosts.values())
+    if len(host_list) < 2:
+        pytest.skip("Dual tests require at least 2 hosts")
+
+    tx_host = host_list[0]
+    rx_host = host_list[1]
+
+    # Create input file on TX host
+    input_file_path = media_create.create_text_file(
+        size_kb=file_size_kb,
+        output_path=os.path.join(media, "test_anc.txt"),
+        host=tx_host,
+    )
+
+    capture_cfg = dict(test_config.get("capture_cfg", {})) if test_config else {}
+    capture_cfg["test_name"] = (
+        f"test_st40p_fps_size_dual_{fps}_{file_size_kb}kb_{framebuff}"
+    )
+
+    try:
+        result = GstreamerApp.execute_dual_st40_test(
+            build=build,
+            tx_nic_port=tx_host.vfs[0],
+            rx_nic_port=rx_host.vfs[0],
+            input_path=input_file_path,
+            payload_type=113,
+            queues=4,
+            tx_framebuff_cnt=framebuff,
+            tx_fps=fps,
+            tx_did=67,
+            tx_sdid=2,
+            timeout=40000,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            capture_cfg=capture_cfg,
+        )
+
+        assert result, f"GStreamer dual ST40P test failed for fps={fps}, size={file_size_kb}KB"
+
+    finally:
+        # Remove the input file on TX host
+        media_create.remove_file(input_file_path, host=tx_host)

tests/validation/tests/dual/gstreamer/__init__.py

@@ -0,0 +1,74 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024-2025 Intel Corporation
+
+import os
+
+import mtl_engine.media_creator as media_create
+import pytest
+from mtl_engine import GstreamerApp
+
+
+@pytest.mark.parametrize("audio_format", ["s8", "s16le", "s24le"])
+@pytest.mark.parametrize("audio_channel", [1, 2])
+@pytest.mark.parametrize("audio_rate", [44100, 48000, 96000])
+def test_audio_format_dual(
+    hosts,
+    build,
+    media,
+    nic_port_list,
+    audio_format,
+    audio_channel,
+    audio_rate,
+    test_time,
+    test_config,
+    prepare_ramdisk,
+):
+    """Test GStreamer ST30 audio format in dual host configuration."""
+    # Get TX and RX hosts
+    host_list = list(hosts.values())
+    if len(host_list) < 2:
+        pytest.skip("Dual tests require at least 2 hosts")
+
+    tx_host = host_list[0]
+    rx_host = host_list[1]
+
+    # Create input file on TX host
+    input_file_path = os.path.join(media, "test_audio.pcm")
+
+    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=tx_host,
+    )
+
+    capture_cfg = dict(test_config.get("capture_cfg", {})) if test_config else {}
+    capture_cfg["test_name"] = (
+        f"test_audio_format_dual_{audio_format}_{audio_channel}_{audio_rate}"
+    )
+
+    try:
+        result = GstreamerApp.execute_dual_st30_test(
+            build=build,
+            tx_nic_port=tx_host.vfs[0],
+            rx_nic_port=rx_host.vfs[0],
+            input_path=input_file_path,
+            payload_type=111,
+            queues=4,
+            audio_format=audio_format,
+            channels=audio_channel,
+            sampling=audio_rate,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            capture_cfg=capture_cfg,
+        )
+
+        assert result, f"GStreamer dual audio format test failed for format {audio_format}"
+
+    finally:
+        # Remove the input file on TX host
+        media_create.remove_file(input_file_path, host=tx_host)

tests/validation/tests/dual/gstreamer/anc_format/__init__.py

@@ -0,0 +1,72 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024-2025 Intel Corporation
+
+import os
+
+import mtl_engine.media_creator as media_create
+import pytest
+from mtl_engine import GstreamerApp
+from mtl_engine.media_files import gstreamer_formats
+
+
+@pytest.mark.parametrize("file", gstreamer_formats.keys())
+def test_video_format_dual(
+    hosts,
+    build,
+    media,
+    nic_port_list,
+    file,
+    test_time,
+    test_config,
+    prepare_ramdisk,
+):
+    """
+    Test GStreamer ST20P video format in dual host configuration using single host setup functions.
+    This test now reuses single host pipeline setup functions with dual host networking updates.
+    """
+    video_file = gstreamer_formats[file]
+
+    # Get TX and RX hosts
+    host_list = list(hosts.values())
+    if len(host_list) < 2:
+        pytest.skip("Dual tests require at least 2 hosts")
+
+    tx_host = host_list[0]
+    rx_host = host_list[1]
+
+    # 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"]),
+        media_path=media,
+        host=tx_host,
+    )
+
+    capture_cfg = dict(test_config.get("capture_cfg", {})) if test_config else {}
+    capture_cfg["test_name"] = f"test_video_format_dual_{file}"
+
+    try:
+        result = GstreamerApp.execute_dual_st20p_test(
+            build=build,
+            tx_nic_port=tx_host.vfs[0],
+            rx_nic_port=rx_host.vfs[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"]),
+            payload_type=112,
+            queues=4,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            capture_cfg=capture_cfg,
+        )
+
+        assert result, f"GStreamer dual video format test failed for format {file}"
+
+    finally:
+        # Remove the input file on TX host
+        media_create.remove_file(input_file_path, host=tx_host)

tests/validation/tests/dual/gstreamer/anc_format/test_anc_format_dual.py

@@ -0,0 +1,84 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024-2025 Intel Corporation
+
+import os
+
+import mtl_engine.media_creator as media_create
+import pytest
+from mtl_engine import GstreamerApp
+from mtl_engine.media_files import yuv_files
+from tests.xfail import SDBQ1971_conversion_v210_720p_error
+
+
+@pytest.mark.parametrize(
+    "file",
+    [
+        pytest.param(f, marks=pytest.mark.smoke) if f == "i1080p59" else f
+        for f in yuv_files.keys()
+    ],
+)
+def test_video_resolutions_dual(
+    hosts,
+    build,
+    media,
+    nic_port_list,
+    file,
+    request,
+    test_time,
+    test_config,
+    prepare_ramdisk,
+):
+    """Test GStreamer ST20P video resolution in dual host configuration."""
+    video_file = yuv_files[file]
+    video_file["format"] = "v210"
+
+    # Get TX and RX hosts
+    host_list = list(hosts.values())
+    if len(host_list) < 2:
+        pytest.skip("Dual tests require at least 2 hosts")
+
+    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,
+    )
+
+    # 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"]),
+        media_path=media,
+        host=tx_host,
+    )
+
+    capture_cfg = dict(test_config.get("capture_cfg", {})) if test_config else {}
+    capture_cfg["test_name"] = f"test_video_resolutions_dual_{file}"
+
+    try:
+        result = GstreamerApp.execute_dual_st20p_test(
+            build=build,
+            tx_nic_port=tx_host.vfs[0],
+            rx_nic_port=rx_host.vfs[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"]),
+            payload_type=112,
+            queues=4,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            capture_cfg=capture_cfg,
+        )
+
+        assert result, f"GStreamer dual video resolution test failed for resolution {file}"
+
+    finally:
+        # Remove the input file on TX host
+        media_create.remove_file(input_file_path, host=tx_host)