@@ -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
@@ -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,
@@ -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:
@@ -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.
new file mode 100644
@@ -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)
@@ -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
@@ -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,
@@ -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)
)
@@ -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(