Making the host ip device configurable

itf/core/base/target/config/base_processor.py

@@ -11,19 +11,20 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 import json
+from typing import Optional
 
 
 class BaseProcessor:
     def __init__(
         self,
         name: str,
-        ip_address: str = None,
+        ip_address: Optional[str] = None,
         ssh_port: int = 22,
-        diagnostic_ip_address: str = None,
-        diagnostic_address: int = None,
-        serial_device: str = None,
+        diagnostic_ip_address: Optional[str] = None,
+        diagnostic_address: Optional[int] = None,
+        serial_device: Optional[str] = None,
         use_doip: bool = False,
-        params: dict = None,
+        params: Optional[dict] = None,
     ):
         """Initialize the BaseProcessor class.
 
@@ -48,7 +49,7 @@ def __init__(
             self.__dict__.update(**params)
 
     @property
-    def name(self):
+    def name(self) -> str:
         return self.__name
 
     @property

itf/core/base/target/config/config.py

@@ -44,8 +44,10 @@ def load_configuration(config_file: str):
                 ip_address=perf_config["ip_address"],
                 ssh_port=perf_config["ssh_port"],
                 ext_ip_address=perf_config["ext_ip_address"],
+                host_ip = perf_config.get("host_ip", None),
                 diagnostic_ip_address=perf_config["diagnostic_ip_address"],
                 diagnostic_address=int(perf_config["diagnostic_address"], 16),
+                diagnostic_host_ip = perf_config.get("diagnostic_host_ip", None),
                 serial_device=perf_config["serial_device"],
                 network_interfaces=perf_config["network_interfaces"],
                 ecu_name=perf_config["ecu_name"],

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

@@ -11,34 +11,39 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 from itf.core.base.target.config.base_processor import BaseProcessor
+from typing import Optional
 
 
 class PerformanceProcessor(BaseProcessor):
     # pylint: disable=dangerous-default-value, too-many-arguments
     def __init__(
         self,
         name: str,
-        ip_address: str = None,
+        ip_address: Optional[str] = None,
         ssh_port: int = 22,
-        ext_ip_address: str = None,
-        diagnostic_ip_address: str = None,
+        ext_ip_address: Optional[str] = None,
+        host_ip: Optional[str] = None,
+        diagnostic_ip_address: Optional[str] = None,
         diagnostic_address: int = 0,
-        serial_device: str = None,
+        diagnostic_host_ip: Optional[str] = None,
+        serial_device: Optional[str] = None,
         network_interfaces: list = [],
-        ecu_name: str = None,
-        data_router_config: dict = None,
+        ecu_name: Optional[str] = None,
+        data_router_config: Optional[dict] = None,
         qemu_num_cores: int = 2,
         qemu_ram_size: str = "1G",
-        params: dict = None,
+        params: Optional[dict] = None,
     ):
         """Initialize the PerformanceProcessor class.
 
         :param str name: The name of the processor.
         :param str ip_address: The IP address of the processor.
         :param int ssh_port: The SSH port for the processor.
         :param str ext_ip_address: The external IP address of the processor.
+        :param str host_ip: The ip of the host, used to find the correct network device.
         :param str diagnostic_ip_address: The internal IP address for diagnostics.
         :param int diagnostic_address: The diagnostic address of the processor.
+        :param str host_diagnosticip: The ip of the host for the diagnostic interface, used to find the correct network device.
         :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.
@@ -58,6 +63,8 @@ def __init__(
             params=params,
         )
         self.__ext_ip_address = ext_ip_address
+        self.__host_ip = host_ip
+        self.__diagnostic_host_ip = diagnostic_host_ip
         self.__network_interfaces = network_interfaces
         self.__ecu_name = ecu_name
         self.__data_router_config = data_router_config
@@ -69,6 +76,14 @@ def __init__(
     def ext_ip_address(self):
         return self.__ext_ip_address
 
+    @property
+    def host_ip(self) -> Optional[str]:
+        return self.__host_ip
+
+    @property
+    def host_diagnostic_ip(self) -> Optional[str]:
+        return self.__diagnostic_host_ip
+
     @property
     def ecu_name(self):
         return self.__ecu_name

itf/core/base/target/hw_target.py

@@ -29,12 +29,17 @@ def hw_target(target_config, test_config):
     diagnostic_ip = None
 
     with nullcontext():
-        with DltReceive(
-            target_ip=target_config.ip_address,
-            protocol=Protocol.UDP,
-            data_router_config=target_config.data_router_config,
-            binary_path=test_config.dlt_receive_path,
-        ):
+        dlt = None
+
+        if target_config.data_router_config:
+            dlt = DltReceive(
+                    target_ip=target_config.ip_address,
+                    protocol=Protocol.UDP,
+                    data_router_config=target_config.data_router_config,
+                    binary_path=test_config.dlt_receive_path,
+                    )
+
+        with dlt if dlt else nullcontext():
             target = Target(test_config.ecu, test_config.os, diagnostic_ip)
             target.register_processors()
             yield target

itf/core/base/target/qemu_target.py

@@ -17,7 +17,7 @@
 from itf.core.base.target.config.ecu import Ecu
 from itf.core.base.target.processors.qemu_processor import TargetProcessorQemu
 from itf.core.dlt.dlt_receive import DltReceive, Protocol
-from itf.core.qemu.qemu_process import QemuProcess as Qemu
+from itf.core.qemu.qemu_process import QemuProcess
 
 
 class TargetQemu(Target):
@@ -38,16 +38,20 @@ def qemu_target(target_config, test_config):
 
     Currently, only ITF tests against an already running Qemu instance is supported.
     """
-    with Qemu(
-        target_config.qemu_image_path, None, target_config.qemu_ram_size, target_config.qemu_num_cores
-    ) if target_config.qemu_image_path else nullcontext() as qemu_process:
-        with DltReceive(
-            target_ip=target_config.ip_address,
-            protocol=Protocol.UDP,
-            data_router_config=target_config.data_router_config,
-            binary_path=test_config.dlt_receive_path,
-        ):
+    with QemuProcess(target_config) if target_config else nullcontext() as qemu_process:
+        dlt = None
+
+        if target_config.data_router_config:
+            dlt = DltReceive(
+                    target_ip=target_config.ip_address,
+                    protocol=Protocol.UDP,
+                    data_router_config=target_config.data_router_config,
+                    binary_path=test_config.dlt_receive_path,
+                    )
+
+        with dlt if dlt else nullcontext():
             target = TargetQemu(test_config.ecu, test_config.os)
             target.register_processors(qemu_process)
             yield target
             target.teardown()
+

itf/core/base/target/qvp_target.py

@@ -42,12 +42,17 @@ def qvp_target(target_config, test_config):
     Currently, only ITF tests against an already running QQVP instance is supported.
     """
     with nullcontext() as qvp_process:
-        with DltReceive(
-            target_ip=target_config.ip_address,
-            protocol=Protocol.UDP,
-            data_router_config=target_config.data_router_config,
-            binary_path=test_config.dlt_receive_path,
-        ):
+        dlt = None
+
+        if target_config.data_router_config:
+            dlt = DltReceive(
+                target_ip=target_config.ip_address,
+                protocol=Protocol.UDP,
+                data_router_config=target_config.data_router_config,
+                binary_path=test_config.dlt_receive_path,
+            )
+
+        with dlt if dlt else nullcontext():
             target = TargetQvp(test_config.ecu, test_config.os)
             target.register_processors(qvp_process)
             yield target

itf/core/qemu/qemu.py

@@ -16,6 +16,9 @@
 import subprocess
 import netifaces
 import logging
+from typing import Optional
+
+from itf.core.base.target.config.ecu import Ecu
 
 logger = logging.getLogger(__name__)
 
@@ -25,42 +28,26 @@ class Qemu:
     This class shall be used to start an qemu instance based on pre-configured Qemu parameters.
     """
 
-    def __init__(
-        self,
-        path_to_image,
-        path_to_bootloader=None,
-        ram="1G",
-        cores="2",
-        cpu="Cascadelake-Server-v5",
-        host_first_network_device_ip_address="160.48.199.77",
-        host_second_network_device_ip_address="192.168.1.99",
-    ):
+    def __init__(self, config: Ecu):
         """Create a QEMU instance with the specified parameters.
-
-        :param str path_to_image: The path to the Qemu image file.
-        :param str path_to_bootloader: The path to the Qemu bootloader file.
-        :param str ram: The amount of RAM to allocate to the QEMU instance.
-        :param str cores: The number of CPU cores to allocate to the QEMU instance.
-        :param str cpu: The CPU model to emulate.
-         Default is Cascadelake-Server-v5 used to emulate modern Intel CPU features.
-         For older Ubuntu versions change that to host in case of errors.
-        :param str host_first_network_device_ip_address: The IP address of the first network device on the host.
-        :param str host_second_network_device_ip_address: The IP address of the second network device on the host.
         """
+
+        self.__config = config
         self.__qemu_path = "/usr/bin/qemu-system-x86_64"
 
         self.__first_network_device_name = "unknown"
         self.__second_network_device_name = "unknown"
         self.__first_network_adapter_mac = "52:54:11:22:33:01"
         self.__second_network_adapter_mac = "52:54:11:22:33:02"
-        self.__first_network_device_ip_address = host_first_network_device_ip_address
-        self.__second_network_device_ip_address = host_second_network_device_ip_address
 
-        self.__path_to_image = path_to_image
-        self.__path_to_bootloader = path_to_bootloader
-        self.__ram = ram
-        self.__cores = cores
-        self.__cpu = cpu
+        self.__first_network_device_ip_address = config.host_ip if config.host_ip else "160.48.199.77"
+        self.__second_network_device_ip_address: Optional[str] = config.host_diagnostic_ip
+
+        self.__path_to_image = config.qemu_image_path
+        self.__path_to_bootloader = None
+        self.__ram = config.qemu_ram_size
+        self.__cores = config.qemu_num_cores
+        self.__cpu = "Cascadelake-Server-v5"
 
         self.__check_qemu_is_installed()
         self.__find_available_kvm_support()
@@ -69,13 +56,13 @@ def __init__(
 
         self._subprocess = None
 
-    def __enter__(self):
+    def __enter__(self) -> subprocess.Popen:
         return self.start()
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.stop()
 
-    def start(self, subprocess_params=None):
+    def start(self, subprocess_params=None) -> subprocess.Popen:
         logger.debug(self.__build_qemu_command())
         subprocess_args = {"args": shlex.split(self.__build_qemu_command())}
         if subprocess_params:
@@ -84,6 +71,9 @@ def start(self, subprocess_params=None):
         return self._subprocess
 
     def stop(self):
+        if self._subprocess is None:
+            raise RuntimeError(f"{self.__class__.__name__}.stop() called before start()")
+
         if self._subprocess.poll() is None:
             self._subprocess.terminate()
             self._subprocess.wait(2)
@@ -130,11 +120,13 @@ def __find_tap_devices(self):
             except KeyError:
                 pass
 
-        if "unknown" in (self.__first_network_device_name, self.__second_network_device_name):
+        if (self.__first_network_device_name == "unknown" and self.__second_network_device_name == "unknown"):
             logger.fatal("Could not find correct tap devices. Please setup network for Qemu first!")
+            logger.fatal(f"Tried to find devices with IPs: {self.__first_network_device_ip_address} {self.__second_network_device_ip_address}")
             sys.exit(-1)
+        logger.info(f"Found devices {self.__first_network_device_ip_address} {self.__second_network_device_ip_address}")
 
-    def __build_qemu_command(self):
+    def __build_qemu_command(self) -> str:
         return (
             f"{self.__qemu_path}"
             " --enable-kvm"  # Use hardware virtualization for better performance
@@ -145,18 +137,21 @@ def __build_qemu_command(self):
             " -nographic"  # Disable graphical display (console-only)
             " -serial mon:stdio"  # Redirect serial output to console
             " -object rng-random,filename=/dev/urandom,id=rng0"  # Provide hardware random number generation
-            f" {self.__first_network_adapter()}"
-            f" {self.__second_network_adapter()}"
+            f"{self.__first_network_adapter()}"
+            f"{self.__second_network_adapter()}"
             " -device virtio-rng-pci,rng=rng0"  # Provide hardware random number generation
         )
 
-    def __first_network_adapter(self):
+    def __first_network_adapter(self) -> str:
         return (
             f" -netdev tap,id=t1,ifname={self.__first_network_device_name},script=no,downscript=no"
             f" -device virtio-net-pci,netdev=t1,id=nic1,mac={self.__first_network_adapter_mac},guest_csum=off"
         )
 
-    def __second_network_adapter(self):
+    def __second_network_adapter(self) -> str:
+        if not self.__config.host_diagnostic_ip:
+            return ""
+
         return (
             f" -netdev tap,id=t2,ifname={self.__second_network_device_name},script=no,downscript=no"
             f" -device virtio-net-pci,netdev=t2,id=nic2,mac={self.__second_network_adapter_mac},guest_csum=off"

itf/core/qemu/qemu_process.py

@@ -15,28 +15,26 @@
 
 from itf.core.utils.process.console import PipeConsole
 from itf.core.qemu.qemu import Qemu
+from itf.core.base.target.config.ecu import Ecu
 
 logger = logging.getLogger(__name__)
 
 
 class QemuProcess:
-    def __init__(self, path_to_qemu_image, path_to_bootloader, available_ram, available_cores):
-        self._path_to_qemu_image = path_to_qemu_image
-        self._path_to_bootloader = path_to_bootloader
-        self._available_ram = available_ram
-        self._available_cores = available_cores
-        self._qemu = Qemu(
-            self._path_to_qemu_image, self._path_to_bootloader, self._available_ram, self._available_cores
-        )
+
+    def __init__(self, config: Ecu):
+
+        self._path_to_qemu_image = config.qemu_image_path
+        self._qemu = Qemu(config)
         self._console = None
 
-    def __enter__(self):
+    def __enter__(self) -> "QemuProcess":
         return self.start()
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.stop()
 
-    def start(self):
+    def start(self) -> "QemuProcess":
         logger.info("Starting Qemu...")
         logger.info(f"Using QEMU image: {self._path_to_qemu_image}")
         subprocess_params = {
@@ -58,5 +56,5 @@ def restart(self):
         self.start()
 
     @property
-    def console(self):
+    def console(self) -> PipeConsole:
         return self._console