|WARNING| pw141743-141745 [PATCH v5 1/4] dts: add context manager for interactive shells

[qemudev@loongson.cn]

Test-Label: loongarch-compilation
Test-Status: WARNING
http://dpdk.org/patch/141743

_apply patch failure_

Submitter: Jeremy Spewock <jspewock@iol.unh.edu>
Date: Tue, 25 Jun 2024 12:27:13 -0400
DPDK git baseline: Repo:dpdk
  Branch: main
  CommitID: 9ab7baa5c14b1ba928c09bda4734827d6d367d6b

Apply patch set 141743-141745 failed:

Checking patch dts/framework/remote_session/__init__.py...
error: while searching for:
from .interactive_shell import InteractiveShell
from .python_shell import PythonShell
from .remote_session import CommandResult, RemoteSession
from .ssh_session import SSHSession
from .testpmd_shell import NicCapability, TestPmdShell


error: patch failed: dts/framework/remote_session/__init__.py:21
error: dts/framework/remote_session/__init__.py: patch does not apply
Checking patch dts/framework/remote_session/interactive_shell.py...
error: while searching for:
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2023 University of New Hampshire

"""Common functionality for interactive shell handling.

The base class, :class:`InteractiveShell`, is meant to be extended by subclasses that contain
functionality specific to that shell type. These subclasses will often modify things like
the prompt to expect or the arguments to pass into the application, but still utilize
the same method for sending a command and collecting output. How this output is handled however
is often application specific. If an application needs elevated privileges to start it is expected
that the method for gaining those privileges is provided when initializing the class.

The :option:`--timeout` command line argument and the :envvar:`DTS_TIMEOUT`
environment variable configure the timeout of getting the output from command execution.
"""

from abc import ABC
from pathlib import PurePath
from typing import Callable, ClassVar

from paramiko import Channel, SSHClient, channel  # type: ignore[import]

from framework.logger import DTSLogger
from framework.settings import SETTINGS


class InteractiveShell(ABC):
    """The base class for managing interactive shells.

    This class shouldn't be instantiated directly, but instead be extended. It contains
    methods for starting interactive shells as well as sending commands to these shells
    and collecting input until reaching a certain prompt. All interactive applications
    will use the same SSH connection, but each will create their own channel on that
    session.
    """

    _interactive_session: SSHClient
    _stdin: channel.ChannelStdinFile
    _stdout: channel.ChannelFile
    _ssh_channel: Channel
    _logger: DTSLogger
    _timeout: float
    _app_args: str

    #: Prompt to expect at the end of output when sending a command.
    #: This is often overridden by subclasses.
    _default_prompt: ClassVar[str] = ""

    #: Extra characters to add to the end of every command
    #: before sending them. This is often overridden by subclasses and is
    #: most commonly an additional newline character.
    _command_extra_chars: ClassVar[str] = ""

    #: Path to the executable to start the interactive application.
    path: ClassVar[PurePath]

    #: Whether this application is a DPDK app. If it is, the build directory
    #: for DPDK on the node will be prepended to the path to the executable.
    dpdk_app: ClassVar[bool] = False

    def __init__(
        self,
        interactive_session: SSHClient,
        logger: DTSLogger,
        get_privileged_command: Callable[[str], str] | None,
        app_args: str = "",
        timeout: float = SETTINGS.timeout,
    ) -> None:
        """Create an SSH channel during initialization.

        Args:
            interactive_session: The SSH session dedicated to interactive shells.
            logger: The logger instance this session will use.
            get_privileged_command: A method for modifying a command to allow it to use
                elevated privileges. If :data:`None`, the application will not be started
                with elevated privileges.
            app_args: The command line arguments to be passed to the application on startup.
            timeout: The timeout used for the SSH channel that is dedicated to this interactive
                shell. This timeout is for collecting output, so if reading from the buffer
                and no output is gathered within the timeout, an exception is thrown.
        """
        self._interactive_session = interactive_session
        self._ssh_channel = self._interactive_session.invoke_shell()
        self._stdin = self._ssh_channel.makefile_stdin("w")
        self._stdout = self._ssh_channel.makefile("r")
        self._ssh_channel.settimeout(timeout)
        self._ssh_channel.set_combine_stderr(True)  # combines stdout and stderr streams
        self._logger = logger
        self._timeout = timeout
        self._app_args = app_args
        self._start_application(get_privileged_command)

    def _start_application(self, get_
error: patch failed: dts/framework/remote_session/interactive_shell.py:1
error: dts/framework/remote_session/interactive_shell.py: patch does not apply
Checking patch dts/framework/remote_session/single_active_interactive_shell.py...
Checking patch dts/framework/remote_session/testpmd_shell.py...
error: while searching for:
from framework.settings import SETTINGS
from framework.utils import StrEnum

from .interactive_shell import InteractiveShell


class TestPmdDevice(object):

error: patch failed: dts/framework/remote_session/testpmd_shell.py:26
error: dts/framework/remote_session/testpmd_shell.py: patch does not apply
Checking patch dts/framework/testbed_model/os_session.py...
error: while searching for:
from framework.remote_session import (
    CommandResult,
    InteractiveRemoteSession,
    InteractiveShell,
    RemoteSession,
    create_interactive_session,
    create_remote_session,
)

error: patch failed: dts/framework/testbed_model/os_session.py:32
error: dts/framework/testbed_model/os_session.py: patch does not apply
Checking patch dts/framework/testbed_model/sut_node.py...
error: while searching for:
        unsupported_capas: set[NicCapability] = set()
        self._logger.debug(f"Checking which capabilities from {capabilities} NIC are supported.")
        testpmd_shell = self.create_interactive_shell(TestPmdShell, privileged=True)
        for capability in capabilities:
            if capability not in supported_capas or capability not in unsupported_capas:
                capability.value(testpmd_shell, supported_capas, unsupported_capas)
        del testpmd_shell
        return supported_capas

    def _set_up_build_target(self, build_target_config: BuildTargetConfiguration) -> None:

error: patch failed: dts/framework/testbed_model/sut_node.py:243
error: dts/framework/testbed_model/sut_node.py: patch does not apply
Checking patch dts/framework/testbed_model/traffic_generator/scapy.py...
error: while searching for:
            PythonShell, timeout=5, privileged=True
        )

        # import libs in remote python console
        for import_statement in SCAPY_RPC_SERVER_IMPORTS:
            self.session.send_command(import_statement)

error: patch failed: dts/framework/testbed_model/traffic_generator/scapy.py:221
error: dts/framework/testbed_model/traffic_generator/scapy.py: patch does not apply
Checking patch dts/tests/TestSuite_pmd_buffer_scatter.py...
error: while searching for:
        Test:
            Start testpmd and run functional test with preset mbsize.
        """
        testpmd = self.sut_node.create_interactive_shell(
            TestPmdShell,
            app_parameters=(
                "--mbcache=200 "

error: patch failed: dts/tests/TestSuite_pmd_buffer_scatter.py:101
error: dts/tests/TestSuite_pmd_buffer_scatter.py: patch does not apply
Checking patch dts/tests/TestSuite_smoke_tests.py...
error: while searching for:
            List all devices found in testpmd and verify the configured devices are among them.
        """
        testpmd_driver = self.sut_node.create_interactive_shell(TestPmdShell, privileged=True)
        dev_list = [str(x) for x in testpmd_driver.get_devices()]
        for nic in self.nics_in_node:
            self.verify(
                nic.pci in dev_list,

error: patch failed: dts/tests/TestSuite_smoke_tests.py:100
error: dts/tests/TestSuite_smoke_tests.py: patch does not apply