Add: Gstreamer st40p tests (#1051)

Add st40p tests,
Add sleep_interval and tx_first to manipulate execution of Tx and Rx,
Add fps_change function to change fps to match values in st20p
gstreamer,
Add file creation function of txt file in media_creator.py,
Add test cases of test_anc for fps, file size and framebuff,

tests/validation/tests/Engine/GstreamerApp.py

@@ -3,6 +3,7 @@
 
 import hashlib
 import os
+import time
 
 from tests.Engine.execute import call, log_fail, log_info, wait
 
@@ -213,26 +214,103 @@ def setup_gstreamer_st30_rx_pipeline(
     return pipeline_command
 
 
+def setup_gstreamer_st40p_tx_pipeline(
+    build: str,
+    nic_port_list: str,
+    input_path: str,
+    tx_payload_type: int,
+    tx_queues: int,
+    tx_framebuff_cnt: int,
+    tx_fps: int,
+    tx_did: int,
+    tx_sdid: int,
+):
+    connection_params = create_connection_params(
+        dev_port=nic_port_list, payload_type=tx_payload_type, udp_port=40000, is_tx=True
+    )
+
+    # st40 tx GStreamer command line
+    pipeline_command = [
+        "gst-launch-1.0",
+        "filesrc",
+        f"location={input_path}",
+        "!",
+        "mtl_st40p_tx",
+        f"tx-queues={tx_queues}",
+        f"tx-framebuff-cnt={tx_framebuff_cnt}",
+        f"tx-fps={tx_fps}",
+        f"tx-did={tx_did}",
+        f"tx-sdid={tx_sdid}",
+    ]
+
+    for key, value in connection_params.items():
+        pipeline_command.append(f"{key}={value}")
+
+    pipeline_command.append(
+        f"--gst-plugin-path={build}/ecosystem/gstreamer_plugin/builddir/"
+    )
+
+    return pipeline_command
+
+
+def setup_gstreamer_st40p_rx_pipeline(
+    build: str,
+    nic_port_list: str,
+    output_path: str,
+    rx_payload_type: int,
+    rx_queues: int,
+    timeout: int,
+):
+    connection_params = create_connection_params(
+        dev_port=nic_port_list,
+        payload_type=rx_payload_type,
+        udp_port=40000,
+        is_tx=False,
+    )
+
+    # st40 rx GStreamer command line
+    pipeline_command = [
+        "gst-launch-1.0",
+        "-v",
+        "mtl_st40_rx",
+        f"rx-queues={rx_queues}",
+        f"timeout={timeout}",
+    ]
+
+    for key, value in connection_params.items():
+        pipeline_command.append(f"{key}={value}")
+
+    pipeline_command.extend(["!", "filesink", f"location={output_path}"])
+
+    pipeline_command.append(
+        f"--gst-plugin-path={build}/ecosystem/gstreamer_plugin/builddir/"
+    )
+
+    return pipeline_command
+
+
 def execute_test(
     build: str,
     tx_command: list,
     rx_command: list,
     input_file: str,
     output_file: str,
     type: str,
-    fps: int = None,
+    sleep_interval: int = 0,
+    tx_first: bool = True,
 ):
 
-    tx_process = call(" ".join(tx_command), cwd=build, timeout=120)
-    rx_process = call(" ".join(rx_command), cwd=build, timeout=120)
+    if tx_first:
+        tx_process = call(" ".join(tx_command), cwd=build, timeout=120)
+        time.sleep(sleep_interval)
+        rx_process = call(" ".join(rx_command), cwd=build, timeout=120)
+    else:
+        rx_process = call(" ".join(rx_command), cwd=build, timeout=120)
+        time.sleep(sleep_interval)
+        tx_process = call(" ".join(tx_command), cwd=build, timeout=120)
 
-    tx_output = wait(tx_process)
+    wait(tx_process)
     wait(rx_process)
-    if type == "st20":
-        tx_result = check_tx_output(fps=fps, output=tx_output.splitlines())
-        # rx_result = check_rx_output(fps=fps, output=rx_output.splitlines())
-        if tx_result is False:
-            return False
 
     file_compare = compare_files(input_file, output_file)
 
@@ -244,36 +322,6 @@ def execute_test(
     return False
 
 
-def check_tx_output(fps: int, output: list) -> bool:
-    tx_fps_result = None
-    for line in output:
-        if "TX_VIDEO_SESSION(0,0:st20sink): fps" in line:
-            tx_fps_result = line
-    if tx_fps_result is not None:
-        for x in range(fps, fps - 3, -1):
-            if f"fps {x}" in tx_fps_result:
-                log_info(f"FPS > {x}")
-                return True
-
-    log_fail("tx session failed")
-    return False
-
-
-def check_rx_output(fps: int, output: list) -> bool:
-    rx_fps_result = None
-    for line in output:
-        if "RX_VIDEO_SESSION(0,0:st20src): fps" in line:
-            rx_fps_result = line
-    if rx_fps_result is not None:
-        for x in range(fps, fps - 2, -1):
-            if f"fps {x}" in line:
-                log_info(f"FPS > {x}")
-                return True
-
-    log_fail("rx session failed")
-    return False
-
-
 def compare_files(input_file, output_file):
     if os.path.exists(input_file) and os.path.exists(output_file):
         input_file_size = os.path.getsize(input_file)
@@ -318,3 +366,16 @@ def audio_format_change(file_format, rx_side: bool = False):
             return 16
         else:
             return 24
+
+
+def fps_change(fps: int):
+    if fps == 23:
+        return 2398
+    if fps == 29:
+        return 2997
+    if fps == 59:
+        return 5994
+    if fps == 119:
+        return 11988
+    else:
+        return fps

tests/validation/tests/Engine/media_creator.py

@@ -2,6 +2,8 @@
 # Copyright(c) 2024-2025 Intel Corporation
 
 import os
+import random
+import string
 import subprocess
 import time
 
@@ -90,6 +92,19 @@ def create_audio_file_sox(
         raise
 
 
+def create_text_file(size_kb: int, output_path: str = "test_anc.txt"):
+    size_bytes = size_kb * 1024
+    chars = string.ascii_letters + string.digits + string.punctuation
+    with open(output_path, "w") as f:
+        while size_bytes > 0:
+            chunk_size = min(size_bytes, 1024)
+            f.write("".join(random.choice(chars) for _ in range(chunk_size)))
+            size_bytes -= chunk_size
+
+    log_info(f"Text file created at {output_path} with size {size_kb} KB")
+    return output_path
+
+
 def remove_file(file_path: str):
     if os.path.exists(file_path):
         os.remove(file_path)

tests/validation/tests/single/gstreamer/anc_format/test_anc_format.py

@@ -0,0 +1,114 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024-2025 Intel Corporation
+
+import os
+
+import pytest
+import tests.Engine.GstreamerApp as gstreamerapp
+import tests.Engine.media_creator as media_create
+
+
+@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(
+    build,
+    media,
+    nic_port_list,
+    file_size_kb,
+    fps,
+    framebuff,
+):
+    input_file_path = media_create.create_text_file(
+        size_kb=file_size_kb, output_path=os.path.join(media, "test_anc.txt")
+    )
+
+    tx_config = gstreamerapp.setup_gstreamer_st40p_tx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[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,
+    )
+
+    rx_config = gstreamerapp.setup_gstreamer_st40p_rx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[1],
+        output_path=os.path.join(media, "output_anc.txt"),
+        rx_payload_type=113,
+        rx_queues=4,
+        timeout=61,
+    )
+
+    try:
+        gstreamerapp.execute_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=os.path.join(media, "output_anc.txt"),
+            type="st40",
+            tx_first=False,
+            sleep_interval=0,
+        )
+    finally:
+        # Remove the files after the test
+        media_create.remove_file(input_file_path)
+        media_create.remove_file(os.path.join(media, "output_anc.txt"))
+
+
+@pytest.mark.parametrize("fps", [60])
+@pytest.mark.parametrize("file_size_kb", [100])
+@pytest.mark.parametrize("framebuff", [1, 3, 6, 12])
+def test_st40p_framebuff(
+    build,
+    media,
+    nic_port_list,
+    file_size_kb,
+    fps,
+    framebuff,
+):
+    input_file_path = media_create.create_text_file(
+        size_kb=file_size_kb, output_path=os.path.join(media, "test_anc.txt")
+    )
+
+    tx_config = gstreamerapp.setup_gstreamer_st40p_tx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[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,
+    )
+
+    rx_config = gstreamerapp.setup_gstreamer_st40p_rx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[1],
+        output_path=os.path.join(media, "output_anc.txt"),
+        rx_payload_type=113,
+        rx_queues=4,
+        timeout=61,
+    )
+
+    try:
+        gstreamerapp.execute_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=os.path.join(media, "output_anc.txt"),
+            type="st40",
+            tx_first=False,
+            sleep_interval=0,
+        )
+    finally:
+        # Remove the files after the test
+        media_create.remove_file(input_file_path)
+        media_create.remove_file(os.path.join(media, "output_anc.txt"))

tests/validation/tests/single/gstreamer/audio_format/test_audio_format.py

@@ -55,6 +55,8 @@ def test_audio_format(
             input_file=input_file_path,
             output_file=os.path.join(media, "output_audio.pcm"),
             type="st30",
+            tx_first=False,
+            sleep_interval=1,
         )
     finally:
         media_create.remove_file(input_file_path)

tests/validation/tests/single/gstreamer/video_format/test_video_format.py

@@ -57,10 +57,11 @@ def test_video_format(
             build=build,
             tx_command=tx_config,
             rx_command=rx_config,
-            fps=video_file["fps"],
             input_file=input_file_path,
             output_file=os.path.join(media, "output_video.yuv"),
             type="st20",
+            tx_first=True,
+            sleep_interval=0,
         )
     finally:
         # Remove the video file after the test

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

@@ -33,7 +33,7 @@ def test_video_resolutions(
         input_path=input_file_path,
         width=video_file["width"],
         height=video_file["height"],
-        framerate=video_file["fps"],
+        framerate=gstreamerapp.fps_change(video_file["fps"]),
         format=gstreamerapp.video_format_change(video_file["format"]),
         tx_payload_type=112,
         tx_queues=4,
@@ -45,7 +45,7 @@ def test_video_resolutions(
         output_path=os.path.join(media, "output_video.yuv"),
         width=video_file["width"],
         height=video_file["height"],
-        framerate=video_file["fps"],
+        framerate=gstreamerapp.fps_change(video_file["fps"]),
         format=gstreamerapp.video_format_change(video_file["format"]),
         rx_payload_type=112,
         rx_queues=4,
@@ -56,10 +56,11 @@ def test_video_resolutions(
             build=build,
             tx_command=tx_config,
             rx_command=rx_config,
-            fps=video_file["fps"],
             input_file=input_file_path,
             output_file=os.path.join(media, "output_video.yuv"),
             type="st20",
+            tx_first=True,
+            sleep_interval=0,
         )
     finally:
         # Remove the video file after the test