Merge branch 'main' into feature/dr-unit-test-design

Author: FScholPer

MODULE.bazel

@@ -131,7 +131,7 @@ bazel_dep(name = "platforms", version = "1.0.0")
 # Documentation dependencies
 #
 ###############################################################################
-bazel_dep(name = "score_docs_as_code", version = "4.0.1")
+bazel_dep(name = "score_docs_as_code", version = "4.0.3")
 
 ################################################################################
 #

score/itf/core/com/ssh.py

@@ -66,7 +66,6 @@ def _load_private_key(pkey_path: str):
             paramiko.RSAKey,
             paramiko.ECDSAKey,
             paramiko.Ed25519Key,
-            paramiko.DSSKey,
         ]
 
         load_errors = []

score/itf/plugins/dlt/__init__.py

@@ -11,11 +11,16 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 import json
+import logging
+from contextlib import contextmanager
+
 import pytest
 
 from score.itf.core.utils.bunch import Bunch
-from score.itf.plugins.core import determine_target_scope
-from score.itf.plugins.dlt.dlt_receive import DltReceive, Protocol
+from score.itf.plugins.dlt.dlt_receive import DltReceive, Protocol, protocol_arguments
+
+
+logger = logging.getLogger(__name__)
 
 
 def pytest_addoption(parser):
@@ -31,6 +36,12 @@ def pytest_addoption(parser):
         required=True,
         help="Path to dlt-receive binary.",
     )
+    parser.addoption(
+        "--dlt-receive-on-target-path",
+        action="store",
+        required=False,
+        help="Path to dlt-receive binary cross-compiled for the target platform.",
+    )
 
 
 @pytest.fixture(scope="session")
@@ -66,3 +77,84 @@ def dlt(dlt_config):
         binary_path=dlt_config.dlt_receive_path,
     ):
         yield
+
+
+_DLT_RECEIVE_REMOTE_PATH = "/tmp/dlt-receive"
+_DLT_OUTPUT_DIR = "/tmp"
+
+
+class DltReceiver:
+    """Thin wrapper around an :class:`AsyncProcess` that also tracks the DLT output file."""
+
+    def __init__(self, proc, dlt_file=None):
+        self._proc = proc
+        self.dlt_file = dlt_file
+
+    def __getattr__(self, name):
+        return getattr(self._proc, name)
+
+
+@pytest.fixture()
+def dlt_on_target(request, target, dlt_config):
+    """Upload ``dlt-receive`` to the target and yield a factory for starting it.
+
+    The factory returns a :class:`DltReceiver` handle that delegates to the
+    underlying :class:`~score.itf.core.process.async_process.AsyncProcess`.
+    All receivers started via the factory are stopped automatically when the
+    fixture tears down.
+
+    Example usage::
+
+        def test_example(target, dlt_on_target):
+            with target.wrap_exec("/usr/bin/dlt-daemon"):
+                with dlt_on_target(Protocol.UDP, multicast_ips=["224.0.0.1"]) as receiver:
+                    # ... send messages ...
+                    pass
+                assert "expected" in receiver.get_output()
+                target.download(receiver.dlt_file, "local_trace.dlt")
+    """
+    # Note: Currently dlt_on_target is only used on docker Linux,
+    # so we default to the host-built binary
+    on_target_path = request.config.getoption("dlt_receive_on_target_path", default=None)
+    local_binary = on_target_path or dlt_config.dlt_receive_path
+
+    target.upload(local_binary, _DLT_RECEIVE_REMOTE_PATH)
+    target.execute(f"chmod +x {_DLT_RECEIVE_REMOTE_PATH}")
+
+    receivers = []
+    _counter = 0
+
+    @contextmanager
+    def start(
+        protocol,
+        host_ip="127.0.0.1",
+        target_ip="127.0.0.1",
+        multicast_ips=None,
+        print_to_stdout=True,
+        output_file=None,
+    ):
+        nonlocal _counter
+        _counter += 1
+        dlt_file = output_file or f"{_DLT_OUTPUT_DIR}/dlt-receive-{_counter}.dlt"
+
+        args = protocol_arguments(protocol, host_ip, target_ip, multicast_ips or [])
+        args += ["-o", dlt_file]
+        if print_to_stdout:
+            args += ["-a", "--stdout-flush"]
+        proc = target.execute_async(_DLT_RECEIVE_REMOTE_PATH, args=args)
+        receiver = DltReceiver(proc, dlt_file=dlt_file)
+        receivers.append(proc)
+        try:
+            yield receiver
+        finally:
+            if proc.is_running():
+                proc.stop()
+
+    yield start
+
+    for proc in receivers:
+        try:
+            if proc.is_running():
+                proc.stop()
+        except Exception:
+            logger.warning("Failed to stop on-target dlt-receive", exc_info=True)

score/itf/plugins/docker.py

@@ -134,9 +134,10 @@ def get_output(self) -> str:
 
 
 class DockerTarget(Target):
-    def __init__(self, container):
+    def __init__(self, container, network=None):
         super().__init__()
         self.container = container
+        self.network = network
         self._client = pypi_docker.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
 
     def __getattr__(self, name):
@@ -248,13 +249,36 @@ def download(self, remote_path: str, local_path: str) -> None:
     def restart(self) -> None:
         self.container.restart()
 
-    def get_ip(self):
-        self.container.reload()
-        return self.container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"]
+    def _network_attr(self, key, network=None):
+        """Return a NetworkSettings attribute for the given Docker network.
 
-    def get_gateway(self):
+        If *network* is ``None`` and the target was created with a dedicated
+        network, that network is used.  Otherwise the value from the first
+        attached network that has a non-empty value for *key* is returned.
+        """
+        if network is None and self.network is not None:
+            network = self.network.name
         self.container.reload()
-        return self.container.attrs["NetworkSettings"]["Networks"]["bridge"]["Gateway"]
+        networks = self.container.attrs["NetworkSettings"]["Networks"]
+        if network is not None:
+            if network not in networks:
+                raise RuntimeError(f"Container {self.container.short_id} is not attached to network '{network}'")
+            return networks[network][key]
+        value = next(
+            (v.get(key) for v in networks.values() if v.get(key, "") != ""),
+            None,
+        )
+        if value is None:
+            raise RuntimeError(f"Container {self.container.short_id} has no {key} on any network")
+        return value
+
+    def get_ip(self, network=None):
+        """Return the container IP on the given Docker network."""
+        return self._network_attr("IPAddress", network)
+
+    def get_gateway(self, network=None):
+        """Return the gateway IP on the given Docker network."""
+        return self._network_attr("Gateway", network)
 
     def ssh(self, username="score", password="score", port=2222):
         return Ssh(target_ip=self.get_ip(), port=port, username=username, password=password)
@@ -322,25 +346,40 @@ def target_init(request, _docker_configuration):
 
     docker_image = request.config.getoption("docker_image")
     client = pypi_docker.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
+
     known_keys = {"command", "init", "environment", "volumes", "shm_size", "detach", "auto_remove"}
     reserved_overrides = {k for k in ("detach", "auto_remove") if k in _docker_configuration}
     if reserved_overrides:
         logger.warning(f"docker_configuration contains reserved keys {reserved_overrides} which will be ignored")
     extra_kwargs = {k: v for k, v in _docker_configuration.items() if k not in known_keys}
-    container = client.containers.run(
-        docker_image,
-        _docker_configuration["command"],
-        detach=True,
-        auto_remove=False,
-        init=_docker_configuration["init"],
-        environment=_docker_configuration["environment"],
-        volumes=_docker_configuration["volumes"],
-        shm_size=_docker_configuration["shm_size"],
-        **extra_kwargs,
+
+    # Create a per-container bridge network so that get_ip() / get_gateway()
+    # return addresses unique to this container.
+    network = client.networks.create(
+        f"score_itf_{os.urandom(8).hex()}",
+        driver="bridge",
     )
+
+    try:
+        container = client.containers.run(
+            docker_image,
+            _docker_configuration["command"],
+            detach=True,
+            auto_remove=False,
+            init=_docker_configuration["init"],
+            environment=_docker_configuration["environment"],
+            volumes=_docker_configuration["volumes"],
+            shm_size=_docker_configuration["shm_size"],
+            network=network.name,
+            **extra_kwargs,
+        )
+    except Exception:
+        network.remove()
+        raise
+
     target = None
     try:
-        target = DockerTarget(container)
+        target = DockerTarget(container, network=network)
         yield target
     finally:
         try:
@@ -352,7 +391,13 @@ def target_init(request, _docker_configuration):
         except Exception:
             logger.warning("Coverage extraction failed", exc_info=True)
         try:
-            container.stop(timeout=1)
+            try:
+                container.stop(timeout=1)
+            finally:
+                # Ensure restart() doesn't accidentally delete the container mid-test.
+                container.remove(force=True)
         finally:
-            # Ensure restart() doesn't accidentally delete the container mid-test.
-            container.remove(force=True)
+            try:
+                network.remove()
+            except Exception:
+                logger.warning(f"Failed to remove network {network.name}", exc_info=True)

score/itf/plugins/qemu/qemu_target.py

@@ -172,6 +172,7 @@ def execute_async(self, binary_path, args=None, cwd="/", **kwargs) -> QemuAsyncP
         try:
             transport = ssh_ctx.get_paramiko_client().get_transport()
             channel = transport.open_session()
+            channel.set_combine_stderr(True)
             inner = (
                 f"[ -r /etc/profile ] && . /etc/profile >/dev/null 2>&1; echo $$; cd {shlex.quote(cwd)} && {command}"
             )

test/BUILD

@@ -50,6 +50,22 @@ py_itf_test(
     ],
 )
 
+py_itf_test(
+    name = "test_dlt_on_target",
+    srcs = [
+        "test_dlt_on_target.py",
+    ],
+    args = [
+        "--docker-image-bootstrap=$(location //test/resources:image_load)",
+        "--docker-image=score_itf_examples:latest",
+    ],
+    data = ["//test/resources:image_load"],
+    plugins = [
+        "//score/itf/plugins:dlt_plugin",
+        "//score/itf/plugins:docker_plugin",
+    ],
+)
+
 py_itf_test(
     name = "test_ssh",
     srcs = [