Test: Gstreamer dual tests (#1244)

Fix GStreamer audio tests by providing the correct formats.
Add the ability for tests to be run remotely, via fixing
the local host to work only on components of the tests.
Add the ability for gstreamer tests to be run with dual
host mode.

---------

Co-authored-by: Dawid Wesierski <dawid.wesierski@intel.com>

Author: Falron98

tests/validation/mtl_engine/GstreamerApp.py

@@ -0,0 +1,95 @@
+# 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,
+    )
+
+    # Create output file path for RX host
+    output_file_path = os.path.join(media, "output_anc_dual.txt")
+
+    # Setup TX pipeline using existing function
+    tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline(
+        build=build,
+        nic_port_list=tx_host.vfs[0],
+        input_path=input_file_path,
+        tx_payload_type=113,
+        tx_queues=4,
+        tx_framebuff_cnt=framebuff,
+        tx_fps=fps,
+        tx_did=67,
+        tx_sdid=2,
+    )
+
+    # Setup RX pipeline using existing function
+    rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline(
+        build=build,
+        nic_port_list=rx_host.vfs[0],
+        output_path=output_file_path,
+        rx_payload_type=113,
+        rx_queues=4,
+        timeout=15,
+    )
+
+    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_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=output_file_path,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            sleep_interval=3,
+            tx_first=False,
+            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 and output file on RX host
+        media_create.remove_file(input_file_path, host=tx_host)
+        media_create.remove_file(output_file_path, host=rx_host)

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

@@ -0,0 +1,104 @@
+# 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", "S16BE", "S24BE"])
+@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,
+    # )
+
+    # Create output file path for RX host
+    output_file_path = os.path.join(
+        media, f"output_audio_dual_{audio_format}_{audio_channel}_{audio_rate}.pcm"
+    )
+
+    # Setup TX pipeline using existing function
+    tx_config = GstreamerApp.setup_gstreamer_st30_tx_pipeline(
+        build=build,
+        nic_port_list=tx_host.vfs[0],
+        input_path=input_file_path,
+        tx_payload_type=111,
+        tx_queues=4,
+        audio_format=audio_format,
+        channels=audio_channel,
+        sampling=audio_rate,
+    )
+
+    # Setup RX pipeline using existing function
+    rx_config = GstreamerApp.setup_gstreamer_st30_rx_pipeline(
+        build=build,
+        nic_port_list=rx_host.vfs[0],
+        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,
+    )
+
+    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_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=output_file_path,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            tx_first=False,
+            sleep_interval=1,
+            capture_cfg=capture_cfg,
+        )
+
+        assert (
+            result
+        ), f"GStreamer dual audio format test failed for format {audio_format}"
+
+    finally:
+        # Remove the output file on RX host
+        # media_create.remove_file(input_file_path, host=tx_host)
+        media_create.remove_file(output_file_path, host=rx_host)

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

@@ -0,0 +1,96 @@
+# 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,
+):
+    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"]),
+        duration=3,
+        media_path=media,
+        host=tx_host,
+    )
+
+    # Create output file path for RX host
+    output_file_path = os.path.join(media, f"output_video_dual_{file}.yuv")
+
+    # Setup TX pipeline using existing function
+    tx_config = GstreamerApp.setup_gstreamer_st20p_tx_pipeline(
+        build=build,
+        nic_port_list=tx_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"]),
+        tx_payload_type=112,
+        tx_queues=4,
+    )
+
+    # Setup RX pipeline using existing function
+    rx_config = GstreamerApp.setup_gstreamer_st20p_rx_pipeline(
+        build=build,
+        nic_port_list=rx_host.vfs[0],
+        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,
+    )
+
+    capture_cfg = dict(test_config.get("capture_cfg", {})) if test_config else {}
+    capture_cfg["test_name"] = f"test_video_format_dual_{file}"
+
+    try:
+        # Use the unified execute_test function for dual host execution
+        result = GstreamerApp.execute_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=output_file_path,
+            test_time=test_time,
+            tx_host=tx_host,
+            rx_host=rx_host,
+            tx_first=False,
+            capture_cfg=capture_cfg,
+        )
+
+        assert result, f"GStreamer dual video format test failed for format {file}"
+
+    finally:
+        # Remove the input file on TX host and output file on RX host
+        media_create.remove_file(input_file_path, host=tx_host)
+        media_create.remove_file(output_file_path, host=rx_host)