[PATCH v2 4/6] dts: add python remote interactive shell

["Juraj Linkeš" <juraj.linkes@pantheon.tech>]

The shell can be used to remotely run any Python code interactively.

Signed-off-by: Juraj Linkeš <juraj.linkes@pantheon.tech>
---
 dts/framework/config/__init__.py              | 28 +-----------
 dts/framework/remote_session/__init__.py      |  2 +-
 dts/framework/remote_session/os_session.py    | 42 +++++++++---------
 .../remote/interactive_shell.py               | 18 +++++---
 .../remote_session/remote/python_shell.py     | 24 +++++++++++
 .../remote_session/remote/testpmd_shell.py    | 33 +++-----------
 dts/framework/testbed_model/node.py           | 35 ++++++++++++++-
 dts/framework/testbed_model/sut_node.py       | 43 ++++++++-----------
 dts/tests/TestSuite_smoke_tests.py            |  6 +--
 9 files changed, 119 insertions(+), 112 deletions(-)
 create mode 100644 dts/framework/remote_session/remote/python_shell.py

diff --git a/dts/framework/config/__init__.py b/dts/framework/config/__init__.py
index 72aa021b97..b5830f6301 100644
--- a/dts/framework/config/__init__.py
+++ b/dts/framework/config/__init__.py
@@ -11,8 +11,7 @@
 import os.path
 import pathlib
 from dataclasses import dataclass
-from enum import Enum, auto, unique
-from pathlib import PurePath
+from enum import auto, unique
 from typing import Any, TypedDict, Union
 
 import warlock  # type: ignore
@@ -331,28 +330,3 @@ def load_config() -> Configuration:
 
 
 CONFIGURATION = load_config()
-
-
-@unique
-class InteractiveApp(Enum):
-    """An enum that represents different supported interactive applications.
-
-    The values in this enum must all be set to objects that have a key called
-    "default_path" where "default_path" represents a PurePath object for the path
-    to the application. This default path will be passed into the handler class
-    for the application so that it can start the application.
-    """
-
-    testpmd = {"default_path": PurePath("app", "dpdk-testpmd")}
-
-    @property
-    def path(self) -> PurePath:
-        """Default path of the application.
-
-        For DPDK apps, this will be appended to the DPDK build directory.
-        """
-        return self.value["default_path"]
-
-    @path.setter
-    def path(self, path: PurePath) -> None:
-        self.value["default_path"] = path
diff --git a/dts/framework/remote_session/__init__.py b/dts/framework/remote_session/__init__.py
index 2c408c2557..1155dd8318 100644
--- a/dts/framework/remote_session/__init__.py
+++ b/dts/framework/remote_session/__init__.py
@@ -17,7 +17,7 @@
 from framework.logger import DTSLOG
 
 from .linux_session import LinuxSession
-from .os_session import OSSession
+from .os_session import InteractiveShellType, OSSession
 from .remote import (
     CommandResult,
     InteractiveRemoteSession,
diff --git a/dts/framework/remote_session/os_session.py b/dts/framework/remote_session/os_session.py
index 633d06eb5d..c17a17a267 100644
--- a/dts/framework/remote_session/os_session.py
+++ b/dts/framework/remote_session/os_session.py
@@ -5,11 +5,11 @@
 from abc import ABC, abstractmethod
 from collections.abc import Iterable
 from pathlib import PurePath
-from typing import Union
+from typing import Type, TypeVar
 
-from framework.config import Architecture, InteractiveApp, NodeConfiguration, NodeInfo
+from framework.config import Architecture, NodeConfiguration, NodeInfo
 from framework.logger import DTSLOG
-from framework.remote_session.remote import InteractiveShell, TestPmdShell
+from framework.remote_session.remote import InteractiveShell
 from framework.settings import SETTINGS
 from framework.testbed_model import LogicalCore
 from framework.testbed_model.hw.port import Port
@@ -23,6 +23,8 @@
     create_remote_session,
 )
 
+InteractiveShellType = TypeVar("InteractiveShellType", bound=InteractiveShell)
+
 
 class OSSession(ABC):
     """
@@ -81,30 +83,26 @@ def send_command(
 
     def create_interactive_shell(
         self,
-        shell_type: InteractiveApp,
-        path_to_app: PurePath,
+        shell_cls: Type[InteractiveShellType],
         eal_parameters: str,
         timeout: float,
-    ) -> Union[InteractiveShell, TestPmdShell]:
+        privileged: bool,
+    ) -> InteractiveShellType:
         """
         See "create_interactive_shell" in SutNode
         """
-        match (shell_type):
-            case InteractiveApp.testpmd:
-                return TestPmdShell(
-                    self.interactive_session.session,
-                    self._logger,
-                    path_to_app,
-                    timeout=timeout,
-                    eal_flags=eal_parameters,
-                )
-            case _:
-                self._logger.info(
-                    f"Unhandled app type {shell_type.name}, defaulting to shell."
-                )
-                return InteractiveShell(
-                    self.interactive_session.session, self._logger, path_to_app, timeout
-                )
+        app_command = (
+            self._get_privileged_command(str(shell_cls.path))
+            if privileged
+            else str(shell_cls.path)
+        )
+        return shell_cls(
+            self.interactive_session.session,
+            self._logger,
+            app_command,
+            eal_parameters,
+            timeout,
+        )
 
     @abstractmethod
     def _get_privileged_command(self, command: str) -> str:
diff --git a/dts/framework/remote_session/remote/interactive_shell.py b/dts/framework/remote_session/remote/interactive_shell.py
index 2cabe9edca..1211d91aa9 100644
--- a/dts/framework/remote_session/remote/interactive_shell.py
+++ b/dts/framework/remote_session/remote/interactive_shell.py
@@ -17,13 +17,18 @@ class InteractiveShell:
     _ssh_channel: Channel
     _logger: DTSLOG
     _timeout: float
-    _path_to_app: PurePath
+    _startup_command: str
+    _app_args: str
+    _default_prompt: str = ""
+    path: PurePath
+    dpdk_app: bool = False
 
     def __init__(
         self,
         interactive_session: SSHClient,
         logger: DTSLOG,
-        path_to_app: PurePath,
+        startup_command: str,
+        app_args: str = "",
         timeout: float = SETTINGS.timeout,
     ) -> None:
         self._interactive_session = interactive_session
@@ -34,16 +39,19 @@ def __init__(
         self._ssh_channel.set_combine_stderr(True)  # combines stdout and stderr streams
         self._logger = logger
         self._timeout = timeout
-        self._path_to_app = path_to_app
+        self._startup_command = startup_command
+        self._app_args = app_args
         self._start_application()
 
     def _start_application(self) -> None:
-        """Starts a new interactive application based on _path_to_app.
+        """Starts a new interactive application based on _startup_command.
 
         This method is often overridden by subclasses as their process for
         starting may look different.
         """
-        self.send_command_get_output(f"{self._path_to_app}", "")
+        self.send_command_get_output(
+            f"{self._startup_command} {self._app_args}", self._default_prompt
+        )
 
     def send_command_get_output(self, command: str, prompt: str) -> str:
         """Send a command and get all output before the expected ending string.
diff --git a/dts/framework/remote_session/remote/python_shell.py b/dts/framework/remote_session/remote/python_shell.py
new file mode 100644
index 0000000000..66d5787c86
--- /dev/null
+++ b/dts/framework/remote_session/remote/python_shell.py
@@ -0,0 +1,24 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2023 PANTHEON.tech s.r.o.
+
+from pathlib import PurePath
+
+from .interactive_shell import InteractiveShell
+
+
+class PythonShell(InteractiveShell):
+    _startup_command: str
+    _default_prompt: str = ">>>"
+    path: PurePath = PurePath("python3")
+
+    def _start_application(self) -> None:
+        self._startup_command = f"{self._startup_command}\n"
+        super()._start_application()
+
+    def send_command(self, command: str, prompt: str = _default_prompt) -> str:
+        """Specific way of handling the command for python
+
+        An extra newline character is consumed in order to force the current line into
+        the stdout buffer.
+        """
+        return self.send_command_get_output(f"{command}\n", prompt)
diff --git a/dts/framework/remote_session/remote/testpmd_shell.py b/dts/framework/remote_session/remote/testpmd_shell.py
index c0261c00f6..1288cfd10c 100644
--- a/dts/framework/remote_session/remote/testpmd_shell.py
+++ b/dts/framework/remote_session/remote/testpmd_shell.py
@@ -3,11 +3,6 @@
 
 from pathlib import PurePath
 
-from paramiko import SSHClient  # type: ignore
-
-from framework.logger import DTSLOG
-from framework.settings import SETTINGS
-
 from .interactive_shell import InteractiveShell
 
 
@@ -22,34 +17,18 @@ def __str__(self) -> str:
 
 
 class TestPmdShell(InteractiveShell):
-    expected_prompt: str = "testpmd>"
+    path: PurePath = PurePath("app", "dpdk-testpmd")
+    dpdk_app: bool = True
+    _default_prompt: str = "testpmd>"
     _eal_flags: str
 
-    def __init__(
-        self,
-        interactive_session: SSHClient,
-        logger: DTSLOG,
-        path_to_testpmd: PurePath,
-        eal_flags: str,
-        timeout: float = SETTINGS.timeout,
-    ) -> None:
-        """Initializes an interactive testpmd session using specified parameters."""
-        self._eal_flags = eal_flags
-
-        super(TestPmdShell, self).__init__(
-            interactive_session,
-            logger=logger,
-            path_to_app=path_to_testpmd,
-            timeout=timeout,
-        )
-
     def _start_application(self) -> None:
-        """Starts a new interactive testpmd shell using _path_to_app."""
+        """Starts a new interactive testpmd shell using _startup_command."""
         self.send_command(
-            f"{self._path_to_app} {self._eal_flags} -- -i",
+            f"{self._startup_command} {self._app_args} -- -i",
         )
 
-    def send_command(self, command: str, prompt: str = expected_prompt) -> str:
+    def send_command(self, command: str, prompt: str = _default_prompt) -> str:
         """Specific way of handling the command for testpmd
 
         An extra newline character is consumed in order to force the current line into
diff --git a/dts/framework/testbed_model/node.py b/dts/framework/testbed_model/node.py
index e09931cedf..f70e4d5ce6 100644
--- a/dts/framework/testbed_model/node.py
+++ b/dts/framework/testbed_model/node.py
@@ -7,7 +7,7 @@
 A node is a generic host that DTS connects to and manages.
 """
 
-from typing import Any, Callable
+from typing import Any, Callable, Type
 
 from framework.config import (
     BuildTargetConfiguration,
@@ -15,7 +15,7 @@
     NodeConfiguration,
 )
 from framework.logger import DTSLOG, getLogger
-from framework.remote_session import OSSession, create_session
+from framework.remote_session import InteractiveShellType, OSSession, create_session
 from framework.settings import SETTINGS
 
 from .hw import (
@@ -138,6 +138,37 @@ def create_session(self, name: str) -> OSSession:
         self._other_sessions.append(connection)
         return connection
 
+    def create_interactive_shell(
+        self,
+        shell_cls: Type[InteractiveShellType],
+        timeout: float = SETTINGS.timeout,
+        privileged: bool = False,
+        app_args: str = "",
+    ) -> InteractiveShellType:
+        """Create a handler for an interactive session.
+
+        Instantiate shell_cls according to the remote OS specifics.
+
+        Args:
+            shell_cls: The class of the shell.
+            timeout: Timeout for reading output from the SSH channel. If you are
+                reading from the buffer and don't receive any data within the timeout
+                it will throw an error.
+            privileged: Whether to run the shell with administrative privileges.
+            app_args: The arguments to be passed to the application.
+        Returns:
+            Instance of the desired interactive application.
+        """
+        if not shell_cls.dpdk_app:
+            shell_cls.path = self.main_session.join_remote_path(shell_cls.path)
+
+        return self.main_session.create_interactive_shell(
+            shell_cls,
+            app_args,
+            timeout,
+            privileged,
+        )
+
     def filter_lcores(
         self,
         filter_specifier: LogicalCoreCount | LogicalCoreList,
diff --git a/dts/framework/testbed_model/sut_node.py b/dts/framework/testbed_model/sut_node.py
index bcad364435..f0b017a383 100644
--- a/dts/framework/testbed_model/sut_node.py
+++ b/dts/framework/testbed_model/sut_node.py
@@ -7,21 +7,15 @@
 import tarfile
 import time
 from pathlib import PurePath
-from typing import Union
+from typing import Type
 
 from framework.config import (
     BuildTargetConfiguration,
     BuildTargetInfo,
-    InteractiveApp,
     NodeInfo,
     SutNodeConfiguration,
 )
-from framework.remote_session import (
-    CommandResult,
-    InteractiveShell,
-    OSSession,
-    TestPmdShell,
-)
+from framework.remote_session import CommandResult, InteractiveShellType, OSSession
 from framework.settings import SETTINGS
 from framework.utils import MesonArgs
 
@@ -359,23 +353,24 @@ def run_dpdk_app(
 
     def create_interactive_shell(
         self,
-        shell_type: InteractiveApp,
+        shell_cls: Type[InteractiveShellType],
         timeout: float = SETTINGS.timeout,
-        eal_parameters: EalParameters | None = None,
-    ) -> Union[InteractiveShell, TestPmdShell]:
-        """Create a handler for an interactive session.
+        privileged: bool = False,
+        eal_parameters: EalParameters | str | None = None,
+    ) -> InteractiveShellType:
+        """Factory method for creating a handler for an interactive session.
 
-        This method is a factory that calls a method in OSSession to create shells for
-        different DPDK applications.
+        Instantiate shell_cls according to the remote OS specifics.
 
         Args:
-            shell_type: Enum value representing the desired application.
+            shell_cls: The class of the shell.
             timeout: Timeout for reading output from the SSH channel. If you are
                 reading from the buffer and don't receive any data within the timeout
                 it will throw an error.
+            privileged: Whether to run the shell with administrative privileges.
             eal_parameters: List of EAL parameters to use to launch the app. If this
-                isn't provided, it will default to calling create_eal_parameters().
-                This is ignored for base "shell" types.
+                isn't provided or an empty string is passed, it will default to calling
+                create_eal_parameters().
         Returns:
             Instance of the desired interactive application.
         """
@@ -383,11 +378,11 @@ def create_interactive_shell(
             eal_parameters = self.create_eal_parameters()
 
         # We need to append the build directory for DPDK apps
-        shell_type.path = self.remote_dpdk_build_dir.joinpath(shell_type.path)
-        default_path = self.main_session.join_remote_path(shell_type.path)
-        return self.main_session.create_interactive_shell(
-            shell_type,
-            default_path,
-            str(eal_parameters),
-            timeout,
+        if shell_cls.dpdk_app:
+            shell_cls.path = self.main_session.join_remote_path(
+                self.remote_dpdk_build_dir, shell_cls.path
+            )
+
+        return super().create_interactive_shell(
+            shell_cls, timeout, privileged, str(eal_parameters)
         )
diff --git a/dts/tests/TestSuite_smoke_tests.py b/dts/tests/TestSuite_smoke_tests.py
index 9cf547205f..e73d015bc7 100644
--- a/dts/tests/TestSuite_smoke_tests.py
+++ b/dts/tests/TestSuite_smoke_tests.py
@@ -3,7 +3,7 @@
 
 import re
 
-from framework.config import InteractiveApp, PortConfig
+from framework.config import PortConfig
 from framework.remote_session import TestPmdDevice, TestPmdShell
 from framework.settings import SETTINGS
 from framework.test_suite import TestSuite
@@ -67,9 +67,7 @@ def test_devices_listed_in_testpmd(self) -> None:
         Test:
             Uses testpmd driver to verify that devices have been found by testpmd.
         """
-        testpmd_driver = self.sut_node.create_interactive_shell(InteractiveApp.testpmd)
-        # We know it should always be a TestPmdShell but mypy doesn't
-        assert isinstance(testpmd_driver, TestPmdShell)
+        testpmd_driver = self.sut_node.create_interactive_shell(TestPmdShell)
         dev_list: list[TestPmdDevice] = testpmd_driver.get_devices()
         for nic in self.nics_in_node:
             self.verify(
-- 
2.34.1