Load data router configuration from target config file

README.md

@@ -56,7 +56,7 @@ Steps:
     ```
   * Run ITF test from `itf` folder:
     ```
-    $ bazel test //:test_ssh_bridge_network
+    $ bazel test //test:test_ssh_bridge_network --test_output=all
     ```
   * Note: If it fails check IP Address of started Qemu with `ifconfig` and update IP addresses in `itf/config/target_config.json` for `S_CORE_ECU_QEMU_BRIDGE_NETWORK`
 * Run ssh test with qemu started with port forwarding
@@ -66,5 +66,5 @@ Steps:
     ```
   * Run ITF test from `itf` folder:
     ```
-    $ bazel test //:test_ssh_port_forwarding
+    $ bazel test //test:test_ssh_port_forwarding --test_output=all
     ```

bazel/py_itf_test.bzl

@@ -39,6 +39,7 @@ def py_itf_test(name, srcs, args = [], data = [], plugins = [], **kwargs):
             requirement("docker"),
             requirement("pytest"),
             requirement("paramiko"),
+            requirement("typing-extensions"),
             "@score_itf//:itf",
         ],
         data = [

config/target_config.json

@@ -8,7 +8,11 @@
             "diagnostic_address": "0x91",
             "serial_device": "",
             "network_interfaces": [],
-            "ecu_name": "s_core_ecu_qemu_bridge_network_pp"
+            "ecu_name": "s_core_ecu_qemu_bridge_network_pp",
+            "data_router_config": {
+                "vlan_address": "127.0.0.1",
+                "multicast_addresses": []
+            }
         },
         "safety_processor": {
             "name": "S_CORE_ECU_QEMU_BRIDGE_NETWORK_SC",
@@ -29,7 +33,11 @@
             "diagnostic_address": "0x91",
             "serial_device": "",
             "network_interfaces": [],
-            "ecu_name": "s_core_ecu_qemu_port_forwarding_pp"
+            "ecu_name": "s_core_ecu_qemu_port_forwarding_pp",
+            "data_router_config": {
+                "vlan_address": "127.0.0.1",
+                "multicast_addresses": []
+            }
         },
         "safety_processor": {
             "name": "CORE_ECU_QEMU_PORT_FORWARDING_SC",

itf/plugins/base/base_plugin.py

@@ -32,7 +32,7 @@ def pytest_addoption(parser):
     parser.addoption(
         "--target_config",
         action="store",
-        default="",
+        default="config/target_config.json",
         help="Path to json file with target configurations.",
     )
     parser.addoption(

itf/plugins/base/os/operating_system.py

@@ -14,6 +14,7 @@
 from itf.plugins.base.os.config import global_os_config as os_config
 
 
+# pylint: disable=no-member
 class OperatingSystem(Enum):
     LINUX = os_config.os.linux
     QNX = os_config.os.qnx

itf/plugins/base/target/config/config.py

@@ -48,6 +48,7 @@ def load_configuration(config_file: str):
                 serial_device=perf_config["serial_device"],
                 network_interfaces=perf_config["network_interfaces"],
                 ecu_name=perf_config["ecu_name"],
+                data_router_config=perf_config["data_router_config"],
                 params=perf_config.get("params", {}),
             )
             PERFORMANCE_PROCESSORS[perf_config["name"]] = performance_processor

itf/plugins/base/target/config/performance_processor.py

@@ -14,7 +14,7 @@
 
 
 class PerformanceProcessor(BaseProcessor):
-    # pylint: disable=dangerous-default-value
+    # pylint: disable=dangerous-default-value, too-many-arguments
     def __init__(
         self,
         name: str,
@@ -25,6 +25,7 @@ def __init__(
         serial_device: str = None,
         network_interfaces: list = [],
         ecu_name: str = None,
+        data_router_config: dict = None,
         params: dict = None,
     ):
         """Initialize the PerformanceProcessor class.
@@ -37,6 +38,8 @@ def __init__(
         :param str serial_device: The serial device for the processor.
         :param list network_interfaces: The network interfaces for the processor.
         :param str ecu_name: The ECU name for the processor.
+        :param dict data_router_configs: Configuration for the data router
+         with keys "vlan_address" and "multicast_addresses".
         :param dict params: Additional parameters for the processor.
         """
         super().__init__(
@@ -50,6 +53,7 @@ def __init__(
         self.__ext_ip_address = ext_ip_address
         self.__network_interfaces = network_interfaces
         self.__ecu_name = ecu_name
+        self.__data_router_config = data_router_config
 
     @property
     def ext_ip_address(self):
@@ -63,6 +67,10 @@ def ecu_name(self):
     def network_interfaces(self):
         return self.__network_interfaces
 
+    @property
+    def data_router_config(self):
+        return self.__data_router_config
+
     def update(self, processor):
         """Update the current processor with another processor's parameters.
 

itf/plugins/base/target/hw_target.py

@@ -32,6 +32,7 @@ def hw_target(target_config, test_config):
         with DltReceive(
             target_ip=target_config.ip_address,
             protocol=Protocol.UDP,
+            data_router_config=target_config.data_router_config,
             binary_path="./itf/plugins/dlt/dlt-receive",
         ):
             target = Target(test_config.ecu, test_config.os, diagnostic_ip)

itf/plugins/base/target/qemu_target.py

@@ -41,11 +41,8 @@ def qemu_target(target_config, test_config):
         with DltReceive(
             target_ip=target_config.ip_address,
             protocol=Protocol.UDP,
+            data_router_config=target_config.data_router_config,
             binary_path="./itf/plugins/dlt/dlt-receive",
-            drconfig={  # TODO make this configurable
-                "vlan_addr": "127.0.0.1",
-                "mcast_addrs": [],
-            },
         ):
             target = TargetQemu(test_config.ecu, test_config.os)
             target.register_processors(qemu_process)

itf/plugins/base/target/qvp_target.py

@@ -45,6 +45,7 @@ def qvp_target(target_config, test_config):
         with DltReceive(
             target_ip=target_config.ip_address,
             protocol=Protocol.UDP,
+            data_router_config=target_config.data_router_config,
             binary_path="./itf/plugins/dlt/dlt-receive",
         ):
             target = TargetQvp(test_config.ecu, test_config.os)

itf/plugins/dlt/dlt_receive.py

@@ -43,7 +43,7 @@ def __init__(
         print_to_stdout: bool = False,
         logger_name: str = None,
         sctf: bool = False,
-        drconfig: dict = None,
+        data_router_config: dict = None,
         binary_path: str = None,
     ):
         """Initialize DltReceive instance.
@@ -55,15 +55,18 @@ def __init__(
         :param bool print_to_stdout: If True, DLT logs will be printed to stdout.
         :param str logger_name: Optional name for the logger. If not provided, defaults to the basename of the binary path.
         :param bool sctf: If True, uses SCTF-specific configurations.
-        :param dict drconfig: Configuration for the data router with keys "vlan_addr" and "mcast_addrs".
+        :param dict data_router_config: Configuration for the data router
+         with keys "vlan_address" and "multicast_addresses".
         :param str binary_path: Path to the DLT receive binary.
         """
         self._target_ip = target_ip
         self._protocol = protocol
         self._dlt_file_name = file_name or f"{get_output_dir()}/dlt_receive.dlt"
 
-        self._drconfig = drconfig
-        self._protocol_opts = DltReceive.protocol_arguments(self._target_ip, self._protocol, sctf, self._drconfig)
+        self._data_router_config = data_router_config
+        self._protocol_opts = DltReceive.protocol_arguments(
+            self._target_ip, self._protocol, sctf, self._data_router_config
+        )
 
         dlt_receive_args = ["-o", self._dlt_file_name] if enable_file_output else []
         dlt_receive_args += self._protocol_opts
@@ -84,15 +87,15 @@ def remove_dlt_file(target_file):
             os.remove(target_file)
 
     @staticmethod
-    def protocol_arguments(target_ip, protocol, sctf, drconfig):
+    def protocol_arguments(target_ip, protocol, sctf, data_router_config):
         dlt_port = "3490"
         proto_specific_opts = []
 
         if protocol == Protocol.TCP:
             proto_specific_opts = ["--tcp", target_ip]
         elif protocol == Protocol.UDP:
-            net_if = target_ip if sctf else drconfig["vlan_addr"]
-            mcasts = drconfig["mcast_addrs"]
+            net_if = target_ip if sctf else data_router_config["vlan_address"]
+            mcasts = data_router_config["multicast_addresses"]
             mcast_ip = [val for pair in zip(["--mcast-ip"] * len(mcasts), mcasts) for val in pair]
             proto_specific_opts = ["--udp"] + mcast_ip + ["--net-if", net_if, "--port", dlt_port]
         else:

itf/plugins/dlt/dlt_window.py

@@ -48,7 +48,7 @@ def __init__(
         sctf: bool = False,
         clear_dlt: bool = True,
         dlt_filter: str = None,
-        drconfig: dict = None,
+        data_router_config: dict = None,
         binary_path: str = None,
     ):
         """Initialize DltWindow with target IP, protocol, and optional parameters.
@@ -61,15 +61,16 @@ def __init__(
         :param bool sctf: If True, uses SCTF configuration.
         :param bool clear_dlt: If True, clears the DLT file at initialization.
         :param str dlt_filter: Filter string for DLT messages.
-        :param dict drconfig: Configuration for the data router with keys "vlan_addr" and "mcast_addrs".
+        :param dict data_router_config: Configuration for the data router
+         with keys "vlan_address" and "multicast_addresses".
         :param str binary_path: Path to the dlt-receive binary.
         """
         self._target_ip = target_ip
         self._protocol = protocol
         self._dlt_file = f"{dlt_file or '/tmp/dlt_window.dlt'}"
         self._captured_logs = []
 
-        self._protocol_opts = DltReceive.protocol_arguments(self._target_ip, self._protocol, sctf, drconfig)
+        self._protocol_opts = DltReceive.protocol_arguments(self._target_ip, self._protocol, sctf, data_router_config)
 
         self._dlt_content = None
         self._queried_counter = 0

requirements.in

@@ -3,3 +3,4 @@
 docker==7.1.0
 pytest==8.4.1
 paramiko==3.5.1
+typing-extensions==4.14.1

requirements_lock.txt

@@ -319,6 +319,10 @@ requests==2.32.3 \
     --hash=sha256:55365417734eb18255590a9ff9eb97e9e1da868d4ccd6402399eaf68af20a760 \
     --hash=sha256:70761cfe03c773ceb22aa2f671b4757976145175cdfca038c02654d061d6dcc6
     # via docker
+typing-extensions==4.14.1 \
+    --hash=sha256:38b39f4aeeab64884ce9f74c94263ef78f3c22467c8724005483154c26648d36 \
+    --hash=sha256:d1e1e3b58374dc93031d6eda2420a48ea44a36c2b4766a4fdeb3710755731d76
+    # via -r requirements.in
 urllib3==2.3.0 \
     --hash=sha256:1cee9ad369867bfdbbb48b7dd50374c0967a0bb7710050facf0dd6911440e3df \
     --hash=sha256:f8c5449b3cf0861679ce7e0503c7b44b5ec981bec0d1d3795a07f1ba96f0204d

test/test_dlt.py

@@ -19,9 +19,9 @@ def test_dlt():
         target_ip="127.0.0.1",
         protocol=Protocol.UDP,
         binary_path="./itf/plugins/dlt/dlt-receive",
-        drconfig={
-            "vlan_addr": "127.0.0.1",
-            "mcast_addrs": [
+        data_router_config={
+            "vlan_address": "127.0.0.1",
+            "multicast_addresses": [
                 "239.255.42.99",
                 "231.255.42.99",
                 "234.255.42.99",