tests: adapt ST40 ancillary tests for netsniff-ng change (#1323)

Commit 3b1d805 (“Test: Implement packet capture fixture using
netsniff-ng tool. (#1283)”) dropped the enable_sudo flag from
execute.py, which caused the ST40 ancillary tests and media helper to
fail when they still passed enable_sudo=True. This commit removes the
stale flag usage and switches the temporary ANC input/output files to
/tmp/ so the tests run without elevated privileges on shared CI hosts.

Author: moleksy

tests/validation/mtl_engine/media_creator.py

@@ -20,13 +20,13 @@ def _terminate_gst_process_after_duration(duration: int, host):
     # Find and terminate the gst-launch process on remote host
     terminate_cmd = "pkill -SIGINT gst-launch-1.0"
     try:
-        run(terminate_cmd, host=host, enable_sudo=True)
+        run(terminate_cmd, host=host)
         logger.info("Sent SIGINT to gst-launch process")
     except Exception:
         # Fallback to SIGTERM
         terminate_cmd = "pkill -SIGTERM gst-launch-1.0"
         try:
-            run(terminate_cmd, host=host, enable_sudo=True)
+            run(terminate_cmd, host=host)
             logger.info("Sent SIGTERM to gst-launch process")
         except Exception:
             pass
@@ -76,7 +76,7 @@ def create_video_file(
 
         # Run the gstreamer command (this will block until process is terminated)
         try:
-            result = run(" ".join(command), host=host, enable_sudo=True)
+            result = run(" ".join(command), host=host)
             # Wait for termination thread to complete
             termination_thread.join(timeout=1)
 
@@ -138,7 +138,7 @@ def create_audio_file_sox(
     logger.info(f"Creating audio file with command: {' '.join(command)}")
 
     try:
-        result = run(" ".join(command), host=host, enable_sudo=True)
+        result = run(" ".join(command), host=host)
         return_code = getattr(result, "returncode", getattr(result, "exit_code", 0))
         if return_code == 0:
             logger.info(f"Audio file created at {output_path}")
@@ -174,7 +174,7 @@ def create_text_file(size_kb: int, output_path: str = "test_anc.txt", host=None)
 
     try:
         for cmd in command_parts:
-            result = run(cmd, host=host, enable_sudo=True)
+            result = run(cmd, host=host)
             return_code = getattr(result, "returncode", getattr(result, "exit_code", 0))
             if return_code != 0:
                 raise subprocess.SubprocessError(
@@ -244,8 +244,8 @@ def create_ancillary_rfc8331_pseudo_file(
 
     try:
         for cmd in command_parts:
-            result = run(cmd, host=host, enable_sudo=True)
-            logging.info("f{cmd}")
+            result = run(cmd, host=host)
+            logging.info(f"{cmd}")
             return_code = getattr(result, "returncode", getattr(result, "exit_code", 0))
             if return_code != 0:
                 raise subprocess.SubprocessError(
@@ -264,11 +264,11 @@ def remove_file(file_path: str, host=None):
     # Always attempt to remove the file
     command = f"rm -f {file_path}"
     try:
-        run(command, host=host, enable_sudo=True)
+        run(command, host=host)
 
         # Check if file still exists after removal attempt
         check_cmd = f"test -f {file_path}"
-        result = run(check_cmd, host=host, enable_sudo=True)
+        result = run(check_cmd, host=host)
         return_code = getattr(result, "returncode", getattr(result, "exit_code", 0))
 
         if return_code != 0:

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

@@ -1,12 +1,13 @@
 # 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
 
+TMP_INPUT_FILE = "/tmp/test_anc.txt"
+TMP_OUTPUT_FILE = "/tmp/output_anc_dual.txt"
+
 
 @pytest.mark.dual
 @pytest.mark.parametrize("fps", [24, 25, 30, 50, 60])
@@ -36,12 +37,12 @@ def test_st40p_fps_size_dual(
     # 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"),
+        output_path=TMP_INPUT_FILE,
         host=tx_host,
     )
 
     # Create output file path for RX host
-    output_file_path = os.path.join(media, "output_anc_dual.txt")
+    output_file_path = TMP_OUTPUT_FILE
 
     # Setup TX pipeline using existing function
     tx_config = GstreamerApp.setup_gstreamer_st40p_tx_pipeline(
@@ -63,6 +64,7 @@ def test_st40p_fps_size_dual(
         output_path=output_file_path,
         rx_payload_type=113,
         rx_queues=4,
+        rx_framebuff_cnt=framebuff,
         timeout=15,
     )
 

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

@@ -1,12 +1,13 @@
 # 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
 
+TMP_INPUT_FILE = "/tmp/test_anc.txt"
+TMP_OUTPUT_FILE = "/tmp/output_anc.txt"
+
 
 @pytest.mark.parametrize("fps", [24, 25, 30, 50, 60, 100, 120])
 @pytest.mark.parametrize("file_size_kb", [10, 100])
@@ -28,7 +29,7 @@ def test_st40p_fps_size(
 
     input_file_path = media_create.create_text_file(
         size_kb=file_size_kb,
-        output_path=os.path.join(media, "test_anc.txt"),
+        output_path=TMP_INPUT_FILE,
         host=host,
     )
 
@@ -47,9 +48,10 @@ def test_st40p_fps_size(
     rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline(
         build=build,
         nic_port_list=host.vfs[1],
-        output_path=os.path.join(media, "output_anc.txt"),
+        output_path=TMP_OUTPUT_FILE,
         rx_payload_type=113,
         rx_queues=4,
+        rx_framebuff_cnt=framebuff,
         timeout=15,
     )
 
@@ -59,7 +61,7 @@ def test_st40p_fps_size(
             tx_command=tx_config,
             rx_command=rx_config,
             input_file=input_file_path,
-            output_file=os.path.join(media, "output_anc.txt"),
+            output_file=TMP_OUTPUT_FILE,
             test_time=test_time,
             host=host,
             tx_first=False,
@@ -68,7 +70,7 @@ def test_st40p_fps_size(
     finally:
         # Remove the files after the test
         media_create.remove_file(input_file_path, host=host)
-        media_create.remove_file(os.path.join(media, "output_anc.txt"), host=host)
+        media_create.remove_file(TMP_OUTPUT_FILE, host=host)
 
 
 @pytest.mark.parametrize("fps", [60])
@@ -94,7 +96,7 @@ def test_st40p_framebuff(
 
     input_file_path = media_create.create_text_file(
         size_kb=file_size_kb,
-        output_path=os.path.join(media, "test_anc.txt"),
+        output_path=TMP_INPUT_FILE,
         host=host,
     )
 
@@ -108,15 +110,15 @@ def test_st40p_framebuff(
         tx_fps=fps,
         tx_did=67,
         tx_sdid=2,
-        tx_rfc8331=True,
     )
 
     rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline(
         build=build,
         nic_port_list=host.vfs[1],
-        output_path=os.path.join(media, "output_anc.txt"),
+        output_path=TMP_OUTPUT_FILE,
         rx_payload_type=113,
         rx_queues=4,
+        rx_framebuff_cnt=framebuff,
         timeout=timeout_period + 10,
     )
 
@@ -126,7 +128,7 @@ def test_st40p_framebuff(
             tx_command=tx_config,
             rx_command=rx_config,
             input_file=input_file_path,
-            output_file=os.path.join(media, "output_anc.txt"),
+            output_file=TMP_OUTPUT_FILE,
             test_time=test_time,
             host=host,
             tx_first=False,
@@ -135,7 +137,7 @@ def test_st40p_framebuff(
     finally:
         # Remove the files after the test
         media_create.remove_file(input_file_path, host=host)
-        media_create.remove_file(os.path.join(media, "output_anc.txt"), host=host)
+        media_create.remove_file(TMP_OUTPUT_FILE, host=host)
 
 
 @pytest.mark.parametrize("fps", [24, 25, 30, 50, 60, 100, 120])
@@ -158,7 +160,7 @@ def test_st40p_format_8331(
 
     input_file_path = media_create.create_ancillary_rfc8331_pseudo_file(
         size_frames=fps * 10,
-        output_path=os.path.join(media, "test_anc.txt"),
+        output_path=TMP_INPUT_FILE,
         host=host,
     )
 
@@ -178,9 +180,10 @@ def test_st40p_format_8331(
     rx_config = GstreamerApp.setup_gstreamer_st40p_rx_pipeline(
         build=build,
         nic_port_list=host.vfs[1],
-        output_path=os.path.join(media, "output_anc.txt"),
+        output_path=TMP_OUTPUT_FILE,
         rx_payload_type=113,
         rx_queues=4,
+        rx_framebuff_cnt=framebuff,
         timeout=timeout_period + 10,
         capture_metadata=True,
     )
@@ -191,7 +194,7 @@ def test_st40p_format_8331(
             tx_command=tx_config,
             rx_command=rx_config,
             input_file=input_file_path,
-            output_file=os.path.join(media, "output_anc.txt"),
+            output_file=TMP_OUTPUT_FILE,
             test_time=test_time,
             host=host,
             tx_first=False,
@@ -200,4 +203,4 @@ def test_st40p_format_8331(
     finally:
         # Remove the files after the test
         media_create.remove_file(input_file_path, host=host)
-        media_create.remove_file(os.path.join(media, "output_anc.txt"), host=host)
+        media_create.remove_file(TMP_OUTPUT_FILE, host=host)