Add: netsniff-ng can be executed in all tests where tcpdump capture was added

Author: MateuszGrabuszynski

tests/validation/mtl_engine/GstreamerApp.py

@@ -6,7 +6,7 @@
 import os
 import time
 
-from mtl_engine.RxTxApp import prepare_tcpdump
+from mtl_engine.RxTxApp import prepare_tcpdump, prepare_netsniff
 
 from .execute import log_fail, run
 
@@ -331,6 +331,7 @@ def execute_test(
     tx_process = None
     rx_process = None
     tcpdump = prepare_tcpdump(capture_cfg, host)
+    netsniff = prepare_netsniff(capture_cfg, host)
 
     try:
         if tx_first:
@@ -383,10 +384,13 @@ def execute_test(
                 background=True,
                 enable_sudo=True,
             )
-        # --- Start tcpdump after pipelines are running ---
+        # --- Start packet capture after pipelines are running ---
         if tcpdump:
             logger.info("Starting tcpdump capture...")
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
+        if netsniff:
+            logger.info("Starting netsniff-ng capture...")
+            netsniff.capture(capture_time=capture_cfg.get("capture_time", test_time))
 
         # Let the test run for the specified duration
         logger.info(f"Running test for {test_time} seconds...")
@@ -444,6 +448,8 @@ def execute_test(
                 pass
         if tcpdump:
             tcpdump.stop()
+        if netsniff:
+            netsniff.stop()
 
     # Compare files for validation
     file_compare = compare_files(input_file, output_file)

tests/validation/mtl_engine/ffmpeg_app.py

@@ -9,7 +9,7 @@
 import time
 
 from mfd_connect import SSHConnection
-from mtl_engine.RxTxApp import prepare_tcpdump
+from mtl_engine.RxTxApp import prepare_tcpdump, prepare_netsniff
 
 from . import rxtxapp_config
 from .execute import log_fail, run
@@ -164,6 +164,7 @@ def execute_test(
     rx_proc = None
     tx_proc = None
     tcpdump = prepare_tcpdump(capture_cfg, host)
+    netsniff = prepare_netsniff(capture_cfg, host)
 
     try:
         # Start RX pipeline first
@@ -190,10 +191,13 @@ def execute_test(
             background=True,
             enable_sudo=True,
         )
-        # Start tcpdump after pipelines are running
+        # Start packet capture when pipelines are running
         if tcpdump:
             logger.info("Starting tcpdump capture...")
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
+        if netsniff:
+            logger.info("Starting netsniff-ng capture...")
+            netsniff.capture(capture_time=capture_cfg.get("capture_time", test_time))
 
         # Let the test run for the specified duration
         logger.info(f"Running test for {test_time} seconds...")
@@ -263,6 +267,8 @@ def execute_test(
                     pass
         if tcpdump:
             tcpdump.stop()
+        if netsniff:
+            netsniff.stop()
     passed = False
     match output_format:
         case "yuv":
@@ -331,6 +337,7 @@ def execute_test_rgb24(
     rx_proc = None
     tx_proc = None
     tcpdump = prepare_tcpdump(capture_cfg, host)
+    netsniff = prepare_netsniff(capture_cfg, host)
 
     try:
         # Start RX pipeline first
@@ -356,10 +363,13 @@ def execute_test_rgb24(
             background=True,
             enable_sudo=True,
         )
-        # Start tcpdump after pipelines are running
+        # Start packet capture when pipelines are running
         if tcpdump:
             logger.info("Starting tcpdump capture...")
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
+        if netsniff:
+            logger.info("Starting netsniff-ng capture...")
+            netsniff.capture(capture_time=capture_cfg.get("capture_time", test_time))
 
         logger.info(
             f"Waiting for RX process to complete (test_time: {test_time} seconds)..."
@@ -437,6 +447,8 @@ def execute_test_rgb24(
                     pass
         if tcpdump:
             tcpdump.stop()
+        if netsniff:
+            netsniff.stop()
     if not check_output_rgb24(rx_output, 1):
         log_fail("rx video sessions failed")
         return False
@@ -506,6 +518,7 @@ def execute_test_rgb24_multiple(
     tx_1_proc = None
     tx_2_proc = None
     tcpdump = prepare_tcpdump(capture_cfg, host)
+    netsniff = prepare_netsniff(capture_cfg, host)
 
     try:
         rx_proc = run(
@@ -538,10 +551,13 @@ def execute_test_rgb24_multiple(
             background=True,
             enable_sudo=True,
         )
-        # Start tcpdump after pipelines are running
+        # Start packet capture when pipelines are running
         if tcpdump:
             logger.info("Starting tcpdump capture...")
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
+        if netsniff:
+            logger.info("Starting netsniff-ng capture...")
+            netsniff.capture(capture_time=capture_cfg.get("capture_time", test_time))
 
         logger.info(f"Waiting for RX process (test_time: {test_time} seconds)...")
         rx_proc.wait()
@@ -611,6 +627,8 @@ def execute_test_rgb24_multiple(
                         pass
         if tcpdump:
             tcpdump.stop()
+        if netsniff:
+            netsniff.stop()
     if not check_output_rgb24(rx_output, 2):
         log_fail("rx video session failed")
         return False

tests/validation/mtl_engine/udp_app.py

@@ -7,7 +7,7 @@
 import re
 
 from mtl_engine import udp_app_config
-from mtl_engine.RxTxApp import prepare_tcpdump
+from mtl_engine.RxTxApp import prepare_tcpdump, prepare_netsniff
 
 from .const import LOG_FOLDER
 from .execute import call, log_fail, wait
@@ -63,20 +63,25 @@ def execute_test_sample(
     server_command = f"./build/app/UfdServerSample --sessions_cnt {sessions_cnt}"
 
     tcpdump = prepare_tcpdump(capture_cfg, host)
+    netsniff = prepare_netsniff(capture_cfg, host)
 
     client_proc = call(client_command, build, test_time, sigint=True, env=client_env)
     server_proc = call(server_command, build, test_time, sigint=True, env=server_env)
 
     try:
-        # Start tcpdump capture after traffic is flowing
+        # Start packet capture when traffic is flowing
         if tcpdump:
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
+        if netsniff:
+            netsniff.capture(capture_time=capture_cfg.get("capture_time", test_time))
         # Wait for both processes to finish
         wait(client_proc)
         wait(server_proc)
     finally:
         if tcpdump:
             tcpdump.stop()
+        if netsniff:
+            netsniff.stop()
 
     if not check_received_packets(client_proc.output) or not check_received_packets(
         server_proc.output
@@ -137,7 +142,7 @@ def execute_test_librist(
     receive_proc = call(receive_command, build, test_time, sigint=True, env=receive_env)
 
     try:
-        # Start tcpdump capture after traffic is flowing
+        # Start packet capture when traffic is flowing
         if tcpdump:
             tcpdump.capture(capture_time=capture_cfg.get("capture_time", test_time))
         # Wait for both processes to finish