Test: Replace hardcoded IPs with dynamic IP pool

Author: staszczuk

.github/workflows/nightly-tests.yml

@@ -27,10 +27,15 @@ jobs:
           python3 -m venv .venv
           source .venv/bin/activate
           pip install -r requirements.txt
+      - name: Create session ID
+        run: |
+          runner_name=${{ runner.name }}
+          echo "SESSION_ID=${runner_name##*-}" >> "$GITHUB_ENV"
       - name: Generate test framework config files
         working-directory: tests/validation/configs
         run: |
           python3 gen_config.py \
+            --session_id ${{ env.SESSION_ID }} \
             --build ${{ secrets.RUNNER_BUILD }} \
             --mtl_path ${{ secrets.RUNNER_MTL_PATH }} \
             --ip_address 127.0.0.1 \

.github/workflows/smoke-tests.yml

@@ -32,10 +32,15 @@ jobs:
           python3 -m venv .venv
           source .venv/bin/activate
           pip install -r requirements.txt
+      - name: Create session ID
+        run: |
+          runner_name=${{ runner.name }}
+          echo "SESSION_ID=${runner_name##*-}" >> "$GITHUB_ENV"
       - name: Generate test framework config files
         working-directory: tests/validation/configs
         run: |
           python3 gen_config.py \
+            --session_id ${{ env.SESSION_ID }} \
             --build ${{ secrets.RUNNER_BUILD }} \
             --mtl_path ${{ secrets.RUNNER_MTL_PATH }} \
             --ip_address 127.0.0.1 \

tests/validation/configs/examples/test_config.yaml

@@ -1,3 +1,4 @@
+session_id: 0
 build: /path/to/mtl
 mtl_path: /path/to/mtl
 media_path: /mnt/media

tests/validation/configs/gen_config.py

@@ -3,8 +3,9 @@
 import yaml
 
 
-def gen_test_config(build: str, mtl_path: str) -> str:
+def gen_test_config(session_id: int, build: str, mtl_path: str) -> str:
     test_config = {
+        "session_id": session_id,
         "build": build,
         "mtl_path": mtl_path,
         "media_path": "/mnt/media",
@@ -62,6 +63,13 @@ def main() -> None:
     parser = argparse.ArgumentParser(
         description="Generate example test and topology configs for the test framework."
     )
+    parser.add_argument(
+        "--session_id",
+        type=int,
+        choices=range(0, 256),
+        required=True,
+        help="specify session ID (0 - 255)",
+    )
     parser.add_argument(
         "--build",
         type=str,
@@ -102,7 +110,11 @@ def main() -> None:
     if args.password == "None" and args.key_path == "None":
         parser.error("one of the arguments --password --key_path is required")
     with open("test_config.yaml", "w") as file:
-        file.write(gen_test_config(build=args.build, mtl_path=args.mtl_path))
+        file.write(
+            gen_test_config(
+                session_id=args.session_id, build=args.build, mtl_path=args.mtl_path
+            )
+        )
     with open("topology_config.yaml", "w") as file:
         file.write(
             gen_topology_config(

tests/validation/conftest.py

@@ -7,7 +7,7 @@
 import os
 import shutil
 import time
-from typing import Dict
+from typing import Any, Dict
 
 import pytest
 from common.mtl_manager.mtlManager import MtlManager
@@ -17,6 +17,7 @@
 from mfd_common_libs.custom_logger import add_logging_level
 from mfd_common_libs.log_levels import TEST_FAIL, TEST_INFO, TEST_PASS
 from mfd_connect.exceptions import ConnectionCalledProcessError
+from mtl_engine import ip_pools
 from mtl_engine.const import FRAMES_CAPTURE, LOG_FOLDER, TESTCMD_LVL
 from mtl_engine.csv_report import (
     csv_add_test,
@@ -381,3 +382,9 @@ def fail_test(stage):
     )
 
     clear_result_note()
+
+
+@pytest.fixture(scope="session", autouse=True)
+def init_ip_address_pools(test_config: dict[Any, Any]) -> None:
+    session_id = int(test_config["session_id"])
+    ip_pools.init(session_id=session_id)

tests/validation/mtl_engine/GstreamerApp.py

@@ -6,6 +6,8 @@
 import re
 import time
 
+from mtl_engine import ip_pools
+
 from .execute import is_process_running, log_fail, run
 
 logger = logging.getLogger(__name__)
@@ -27,8 +29,8 @@ def capture_stdout(process, process_name):
 def create_connection_params(
     dev_port: str,
     payload_type: str,
-    dev_ip: str = "192.168.96.3",
-    ip: str = "192.168.96.2",
+    dev_ip: str,
+    ip: str,
     udp_port: int = 20000,
     is_tx: bool = True,
 ) -> dict:
@@ -111,7 +113,12 @@ def setup_gstreamer_st20p_tx_pipeline(
     tx_fps: int = None,
 ):
     connection_params = create_connection_params(
-        dev_port=nic_port_list, payload_type=tx_payload_type, udp_port=20000, is_tx=True
+        dev_port=nic_port_list,
+        payload_type=tx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
+        udp_port=20000,
+        is_tx=True,
     )
 
     framerate = fract_format(framerate)
@@ -168,6 +175,8 @@ def setup_gstreamer_st20p_rx_pipeline(
     connection_params = create_connection_params(
         dev_port=nic_port_list,
         payload_type=rx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
         udp_port=20000,
         is_tx=False,
     )
@@ -210,7 +219,12 @@ def setup_gstreamer_st30_tx_pipeline(
     sampling: int,
 ):
     connection_params = create_connection_params(
-        dev_port=nic_port_list, payload_type=tx_payload_type, udp_port=30000, is_tx=True
+        dev_port=nic_port_list,
+        payload_type=tx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
+        udp_port=30000,
+        is_tx=True,
     )
 
     # st30 tx GStreamer command line
@@ -245,6 +259,8 @@ def setup_gstreamer_st30_rx_pipeline(
     connection_params = create_connection_params(
         dev_port=nic_port_list,
         payload_type=rx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
         udp_port=30000,
         is_tx=False,
     )
@@ -287,7 +303,12 @@ def setup_gstreamer_st40p_tx_pipeline(
     tx_user_controlled_pacing_offset: int = 0,
 ):
     connection_params = create_connection_params(
-        dev_port=nic_port_list, payload_type=tx_payload_type, udp_port=40000, is_tx=True
+        dev_port=nic_port_list,
+        payload_type=tx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
+        udp_port=40000,
+        is_tx=True,
     )
 
     pipeline_command = ["gst-launch-1.0"]
@@ -336,6 +357,8 @@ def setup_gstreamer_st40p_rx_pipeline(
     connection_params = create_connection_params(
         dev_port=nic_port_list,
         payload_type=rx_payload_type,
+        dev_ip=ip_pools.tx[0],
+        ip=ip_pools.rx[0],
         udp_port=40000,
         is_tx=False,
     )

tests/validation/mtl_engine/RxTxApp.py

@@ -6,10 +6,10 @@
 import os
 import re
 import subprocess
-import sys
 import time
 
 from mfd_connect import SSHConnection
+from mtl_engine import ip_pools
 
 from . import rxtxapp_config
 from .execute import log_fail, run
@@ -18,28 +18,6 @@
 logger = logging.getLogger(__name__)
 
 
-unicast_ip_dict = dict(
-    tx_interfaces="192.168.17.101",
-    rx_interfaces="192.168.17.102",
-    tx_sessions="192.168.17.102",
-    rx_sessions="192.168.17.101",
-)
-
-multicast_ip_dict = dict(
-    tx_interfaces="192.168.17.101",
-    rx_interfaces="192.168.17.102",
-    tx_sessions="239.168.48.9",
-    rx_sessions="239.168.48.9",
-)
-
-kernel_ip_dict = dict(
-    tx_interfaces="192.168.17.101",
-    rx_interfaces="192.168.17.102",
-    tx_sessions="127.0.0.1",
-    rx_sessions="127.0.0.1",
-)
-
-
 def capture_stdout(proc, proc_name: str):
     """Capture and log stdout from a process"""
     try:
@@ -74,18 +52,18 @@ def add_interfaces(config: dict, nic_port_list: list, test_mode: str) -> dict:
     config["interfaces"][1]["name"] = nic_port_list[1]
 
     if test_mode == "unicast":
-        config["interfaces"][0]["ip"] = unicast_ip_dict["tx_interfaces"]
-        config["interfaces"][1]["ip"] = unicast_ip_dict["rx_interfaces"]
-        config["tx_sessions"][0]["dip"][0] = unicast_ip_dict["tx_sessions"]
-        config["rx_sessions"][0]["ip"][0] = unicast_ip_dict["rx_sessions"]
+        config["interfaces"][0]["ip"] = ip_pools.tx[0]
+        config["interfaces"][1]["ip"] = ip_pools.rx[0]
+        config["tx_sessions"][0]["dip"][0] = ip_pools.rx[0]
+        config["rx_sessions"][0]["ip"][0] = ip_pools.tx[0]
     elif test_mode == "multicast":
-        config["interfaces"][0]["ip"] = multicast_ip_dict["tx_interfaces"]
-        config["interfaces"][1]["ip"] = multicast_ip_dict["rx_interfaces"]
-        config["tx_sessions"][0]["dip"][0] = multicast_ip_dict["tx_sessions"]
-        config["rx_sessions"][0]["ip"][0] = multicast_ip_dict["rx_sessions"]
+        config["interfaces"][0]["ip"] = ip_pools.tx[0]
+        config["interfaces"][1]["ip"] = ip_pools.rx[0]
+        config["tx_sessions"][0]["dip"][0] = ip_pools.rx_multicast[0]
+        config["rx_sessions"][0]["ip"][0] = ip_pools.rx_multicast[0]
     elif test_mode == "kernel":
-        config["tx_sessions"][0]["dip"][0] = kernel_ip_dict["tx_sessions"]
-        config["rx_sessions"][0]["ip"][0] = kernel_ip_dict["rx_sessions"]
+        config["tx_sessions"][0]["dip"][0] = "127.0.0.1"
+        config["rx_sessions"][0]["ip"][0] = "127.0.0.1"
     else:
         log_fail(f"wrong test_mode {test_mode}")
 
@@ -1161,22 +1139,18 @@ def add_dual_interfaces(
     rx_config["interfaces"][0]["name"] = rx_nic_port_list[0]
 
     if test_mode == "unicast":
-        tx_config["interfaces"][0]["ip"] = unicast_ip_dict["tx_interfaces"]
-        rx_config["interfaces"][0]["ip"] = unicast_ip_dict["rx_interfaces"]
-        tx_config["tx_sessions"][0]["dip"][0] = unicast_ip_dict[
-            "rx_interfaces"
-        ]  # TX sends to RX IP
-        rx_config["rx_sessions"][0]["ip"][0] = unicast_ip_dict[
-            "tx_interfaces"
-        ]  # RX listens for TX IP
+        tx_config["interfaces"][0]["ip"] = ip_pools.tx[0]
+        rx_config["interfaces"][0]["ip"] = ip_pools.rx[0]
+        tx_config["tx_sessions"][0]["dip"][0] = ip_pools.rx[0]
+        rx_config["rx_sessions"][0]["ip"][0] = ip_pools.tx[0]
     elif test_mode == "multicast":
-        tx_config["interfaces"][0]["ip"] = multicast_ip_dict["tx_interfaces"]
-        rx_config["interfaces"][0]["ip"] = multicast_ip_dict["rx_interfaces"]
-        tx_config["tx_sessions"][0]["dip"][0] = multicast_ip_dict["tx_sessions"]
-        rx_config["rx_sessions"][0]["ip"][0] = multicast_ip_dict["rx_sessions"]
+        tx_config["interfaces"][0]["ip"] = ip_pools.tx[0]
+        rx_config["interfaces"][0]["ip"] = ip_pools.rx[0]
+        tx_config["tx_sessions"][0]["dip"][0] = ip_pools.rx_multicast[0]
+        rx_config["rx_sessions"][0]["ip"][0] = ip_pools.rx_multicast[0]
     elif test_mode == "kernel":
-        tx_config["tx_sessions"][0]["dip"][0] = kernel_ip_dict["tx_sessions"]
-        rx_config["rx_sessions"][0]["ip"][0] = kernel_ip_dict["rx_sessions"]
+        tx_config["tx_sessions"][0]["dip"][0] = "127.0.0.1"
+        rx_config["rx_sessions"][0]["ip"][0] = "127.0.0.1"
     else:
         log_fail(f"wrong test_mode {test_mode}")