ITF target, ssh and sftp

bazel/py_itf_test.bzl

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

itf/plugins/base/base_plugin.py

@@ -11,11 +11,16 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 import logging
+import socket
 import pytest
 
 from itf.plugins.base.constants import TEST_CONFIG_KEY, TARGET_CONFIG_KEY
 from itf.plugins.base.target.config import load_configuration, target_ecu_argparse
 from itf.plugins.base.os.operating_system import OperatingSystem
+from itf.plugins.base.target.qemu_target import qemu_target
+from itf.plugins.base.target.qvp_target import qvp_target
+from itf.plugins.base.target.hw_target import hw_target
+from itf.plugins.base.test.utils import pre_tests_phase, post_tests_phase
 from itf.plugins.utils import padder
 from itf.plugins.xtf_common.bunch import Bunch
 
@@ -72,6 +77,38 @@ def target_config_fixture(request):
     yield target_config
 
 
+@pytest.fixture(scope="session")
+def target_fixture(target_config_fixture, test_config_fixture, request):
+    logger.info("Starting target_fixture in base_plugin.py ...")
+    logger.info(f"Starting tests on host: {socket.gethostname()}")
+
+    if test_config_fixture.qemu:
+        with qemu_target(test_config_fixture) as qemu:
+            try:
+                pre_tests_phase(qemu, target_config_fixture.ip_address, test_config_fixture, request)
+                yield qemu
+            finally:
+                post_tests_phase(qemu, test_config_fixture)
+
+    elif test_config_fixture.qvp:
+        with qvp_target(test_config_fixture) as qvp:
+            try:
+                pre_tests_phase(qvp, target_config_fixture.ip_address, test_config_fixture, request)
+                yield qvp
+            finally:
+                post_tests_phase(qvp, test_config_fixture)
+
+    elif test_config_fixture.hw:
+        with hw_target(test_config_fixture) as hardware:
+            try:
+                pre_tests_phase(hardware, target_config_fixture.ip_address, test_config_fixture, request)
+                yield hardware
+            finally:
+                post_tests_phase(hardware, test_config_fixture)
+    else:
+        raise RuntimeError("QEMU, QVP or HW not specified to use")
+
+
 def __make_test_config(config):
     load_configuration(config.getoption("target_config"))
     return Bunch(

itf/plugins/base/target/base_target.py

@@ -0,0 +1,112 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+import logging
+
+from itf.plugins.base.target.processors.target_processor import TargetProcessor
+from itf.plugins.base.target.processors.safety_processor import TargetSafetyProcessor
+from itf.plugins.base.os.operating_system import OperatingSystem
+from itf.plugins.base.target.config.ecu import Ecu
+
+logger = logging.getLogger(__name__)
+
+
+class Target:
+    """Represents a set of Processors"""
+
+    def __init__(
+        self,
+        target_ecu: Ecu,
+        target_sut_os: OperatingSystem = OperatingSystem.LINUX,
+        diagnostic_ip: str = None,
+    ):
+        """Initializes the Target with the given parameters.
+
+        :param Ecu target_ecu: The ECU type for the target.
+        :param OperatingSystem target_sut_os: The operating system of the target SUT. Default is LINUX.
+        :param str diagnostic_ip: The IP address for diagnostic communication.
+        """
+        self.__target_ecu = target_ecu
+        self.__target_sut_os = target_sut_os
+        self.__sut = None
+        self.__safety_core = None
+        # Other processors
+        for other_ecu in self.target_ecu.others:
+            setattr(self, other_ecu.name.lower(), None)  # Will be set when registering processors
+        self.__processors = []
+        self.__diagnostic_ip = diagnostic_ip
+
+    def __repr__(self):
+        return str(self.__dict__)
+
+    def __str__(self):
+        return str(self.__dict__)
+
+    # pylint: disable=unused-argument
+    def register_processors(self, process=None, initialize_serial_device=True, initialize_serial_logs=True):
+        self.__sut = TargetProcessor(
+            self.__target_ecu.sut,
+            self.__target_sut_os,
+            self.__diagnostic_ip,
+        )
+        self.__processors.append(self.sut)
+
+        self.__safety_core = TargetSafetyProcessor(
+            self.__target_ecu.sc,
+            OperatingSystem.UNSPECIFIED,
+            diagnostic_ip=self.__diagnostic_ip,
+        )
+
+        for processor in self.target_ecu.others:
+            other_processor = TargetProcessor(processor, OperatingSystem.UNSPECIFIED)
+            self.__processors.append(other_processor)
+            setattr(self, processor.name.lower(), other_processor)
+
+    @property
+    def target_ecu(self):
+        return self.__target_ecu
+
+    @property
+    def target_sut_os(self):
+        return self.__target_sut_os
+
+    @property
+    def diagnostic_ip(self):
+        return self.__diagnostic_ip
+
+    @property
+    def sut(self):
+        return self.__sut
+
+    @sut.setter
+    def sut(self, value):
+        self.__sut = value
+
+    @property
+    def safety_core(self):
+        return self.__safety_core
+
+    @safety_core.setter
+    def safety_core(self, value):
+        self.__safety_core = value
+
+    @property
+    def processors(self):
+        return self.__processors
+
+    @processors.setter
+    def processors(self, value):
+        self.__processors = value
+
+    def teardown(self):
+        for processor in self.__processors:
+            processor.teardown()

itf/plugins/base/target/hw_target.py

@@ -0,0 +1,34 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+import logging
+
+from contextlib import contextmanager, nullcontext
+from itf.plugins.base.target.base_target import Target
+
+
+logger = logging.getLogger(__name__)
+
+
+@contextmanager
+def hw_target(test_config):
+    """Context manager for hardware target setup.
+
+    Currently, only ITF tests against an already running hardware instance is supported.
+    """
+    diagnostic_ip = None
+
+    with nullcontext():
+        target = Target(test_config.ecu, test_config.os, diagnostic_ip)
+        target.register_processors()
+        yield target
+        target.teardown()

itf/plugins/base/target/processors/__init__.py

@@ -0,0 +1,12 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************

itf/plugins/base/target/processors/qemu_processor.py

@@ -0,0 +1,29 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+from itf.plugins.base.os.operating_system import OperatingSystem
+from itf.plugins.base.target.config.base_processor import BaseProcessor
+from itf.plugins.base.target.processors.target_processor import TargetProcessor
+
+
+class TargetProcessorQemu(TargetProcessor):
+    """Target Processor for QEMU."""
+
+    def __init__(self, processor: BaseProcessor, os: OperatingSystem, process):
+        self.__process = process
+        super().__init__(processor, os)
+
+    def kill_process(self):
+        self.__process.stop()
+
+    def restart_process(self):
+        self.__process.restart()

itf/plugins/base/target/processors/qvp_processor.py

@@ -0,0 +1,27 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+from itf.plugins.base.os.operating_system import OperatingSystem
+from itf.plugins.base.target.config.base_processor import BaseProcessor
+from itf.plugins.base.target.processors.target_processor import TargetProcessor
+
+
+class TargetProcessorQVP(TargetProcessor):
+    def __init__(self, processor: BaseProcessor, os: OperatingSystem, process):
+        self._process = process
+        super().__init__(processor, os)
+
+    def kill_process(self):
+        self._process.stop()
+
+    def restart_process(self):
+        pass

itf/plugins/base/target/processors/safety_processor.py

@@ -0,0 +1,23 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+from itf.plugins.base.os.operating_system import OperatingSystem
+from itf.plugins.base.target.config.base_processor import BaseProcessor
+from itf.plugins.base.target.processors.target_processor import TargetProcessor
+
+
+class TargetSafetyProcessor(TargetProcessor):
+    """Represents the Safety processor of the target ECU."""
+
+    # pylint: disable=useless-super-delegation
+    def __init__(self, processor: BaseProcessor, os: OperatingSystem, diagnostic_ip=None):
+        super().__init__(processor, os, diagnostic_ip)