[PATCH v3 7/8] dts: rework interactive shells

[Luca Vizzarro <luca.vizzarro@arm.com>]

The way nodes and interactive shells interact makes it difficult to
develop for static type checking and hinting. The current system relies
on a top-down approach, attempting to give a generic interface to the
test developer, hiding the interaction of concrete shell classes as much
as possible. When working with strong typing this approach is not ideal,
as Python's implementation of generics is still rudimentary.

This rework reverses the tests interaction to a bottom-up approach,
allowing the test developer to call concrete shell classes directly,
and let them ingest nodes independently. While also re-enforcing type
checking and making the code easier to read.

Signed-off-by: Luca Vizzarro <luca.vizzarro@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepanek@arm.com>
---
 dts/framework/params/eal.py                   |   6 +-
 dts/framework/remote_session/dpdk_shell.py    | 104 ++++++++++++++++
 .../remote_session/interactive_shell.py       |  74 +++++++-----
 dts/framework/remote_session/python_shell.py  |   4 +-
 dts/framework/remote_session/testpmd_shell.py |  64 +++++-----
 dts/framework/testbed_model/node.py           |  36 +-----
 dts/framework/testbed_model/os_session.py     |  36 +-----
 dts/framework/testbed_model/sut_node.py       | 112 +-----------------
 .../testbed_model/traffic_generator/scapy.py  |   4 +-
 dts/tests/TestSuite_hello_world.py            |   7 +-
 dts/tests/TestSuite_pmd_buffer_scatter.py     |  21 ++--
 dts/tests/TestSuite_smoke_tests.py            |   2 +-
 12 files changed, 200 insertions(+), 270 deletions(-)
 create mode 100644 dts/framework/remote_session/dpdk_shell.py

diff --git a/dts/framework/params/eal.py b/dts/framework/params/eal.py
index bbdbc8f334..8d7766fefc 100644
--- a/dts/framework/params/eal.py
+++ b/dts/framework/params/eal.py
@@ -35,9 +35,9 @@ class EalParams(Params):
                 ``other_eal_param='--single-file-segments'``
     """
 
-    lcore_list: LogicalCoreList = field(metadata=Params.short("l"))
-    memory_channels: int = field(metadata=Params.short("n"))
-    prefix: str = field(metadata=Params.long("file-prefix"))
+    lcore_list: LogicalCoreList | None = field(default=None, metadata=Params.short("l"))
+    memory_channels: int | None = field(default=None, metadata=Params.short("n"))
+    prefix: str = field(default="dpdk", metadata=Params.long("file-prefix"))
     no_pci: Switch = None
     vdevs: list[VirtualDevice] | None = field(
         default=None, metadata=Params.multiple() | Params.long("vdev")
diff --git a/dts/framework/remote_session/dpdk_shell.py b/dts/framework/remote_session/dpdk_shell.py
new file mode 100644
index 0000000000..25e3df4eaa
--- /dev/null
+++ b/dts/framework/remote_session/dpdk_shell.py
@@ -0,0 +1,104 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2024 Arm Limited
+
+"""DPDK-based interactive shell.
+
+Provides a base class to create interactive shells based on DPDK.
+"""
+
+
+from abc import ABC
+
+from framework.params.eal import EalParams
+from framework.remote_session.interactive_shell import InteractiveShell
+from framework.settings import SETTINGS
+from framework.testbed_model.cpu import LogicalCoreCount, LogicalCoreList
+from framework.testbed_model.sut_node import SutNode
+
+
+def compute_eal_params(
+    node: SutNode,
+    params: EalParams | None = None,
+    lcore_filter_specifier: LogicalCoreCount | LogicalCoreList = LogicalCoreCount(),
+    ascending_cores: bool = True,
+    append_prefix_timestamp: bool = True,
+) -> EalParams:
+    """Compute EAL parameters based on the node's specifications.
+
+    Args:
+        node: The SUT node to compute the values for.
+        params: The EalParams object to amend, if set to None a new object is created and returned.
+        lcore_filter_specifier: A number of lcores/cores/sockets to use
+            or a list of lcore ids to use.
+            The default will select one lcore for each of two cores
+            on one socket, in ascending order of core ids.
+        ascending_cores: Sort cores in ascending order (lowest to highest IDs).
+            If :data:`False`, sort in descending order.
+        append_prefix_timestamp: If :data:`True`, will append a timestamp to DPDK file prefix.
+    """
+    if params is None:
+        params = EalParams()
+
+    if params.lcore_list is None:
+        params.lcore_list = LogicalCoreList(
+            node.filter_lcores(lcore_filter_specifier, ascending_cores)
+        )
+
+    prefix = params.prefix
+    if append_prefix_timestamp:
+        prefix = f"{prefix}_{node._dpdk_timestamp}"
+    prefix = node.main_session.get_dpdk_file_prefix(prefix)
+    if prefix:
+        node._dpdk_prefix_list.append(prefix)
+    params.prefix = prefix
+
+    if params.ports is None:
+        params.ports = node.ports
+
+    return params
+
+
+class DPDKShell(InteractiveShell, ABC):
+    """The base class for managing DPDK-based interactive shells.
+
+    This class shouldn't be instantiated directly, but instead be extended.
+    It automatically injects computed EAL parameters based on the node in the
+    supplied app parameters.
+    """
+
+    _node: SutNode
+    _app_params: EalParams
+
+    _lcore_filter_specifier: LogicalCoreCount | LogicalCoreList
+    _ascending_cores: bool
+    _append_prefix_timestamp: bool
+
+    def __init__(
+        self,
+        node: SutNode,
+        app_params: EalParams,
+        privileged: bool = True,
+        timeout: float = SETTINGS.timeout,
+        lcore_filter_specifier: LogicalCoreCount | LogicalCoreList = LogicalCoreCount(),
+        ascending_cores: bool = True,
+        append_prefix_timestamp: bool = True,
+        start_on_init: bool = True,
+    ) -> None:
+        """Overrides :meth:`~.interactive_shell.InteractiveShell.__init__`."""
+        self._lcore_filter_specifier = lcore_filter_specifier
+        self._ascending_cores = ascending_cores
+        self._append_prefix_timestamp = append_prefix_timestamp
+
+        super().__init__(node, app_params, privileged, timeout, start_on_init)
+
+    def _post_init(self):
+        """Computes EAL params based on the node capabilities before start."""
+        self._app_params = compute_eal_params(
+            self._node,
+            self._app_params,
+            self._lcore_filter_specifier,
+            self._ascending_cores,
+            self._append_prefix_timestamp,
+        )
+
+        self._update_path(self._node.remote_dpdk_build_dir.joinpath(self.path))
diff --git a/dts/framework/remote_session/interactive_shell.py b/dts/framework/remote_session/interactive_shell.py
index 9da66d1c7e..4be7966672 100644
--- a/dts/framework/remote_session/interactive_shell.py
+++ b/dts/framework/remote_session/interactive_shell.py
@@ -17,13 +17,14 @@
 
 from abc import ABC
 from pathlib import PurePath
-from typing import Callable, ClassVar
+from typing import ClassVar
 
-from paramiko import Channel, SSHClient, channel  # type: ignore[import-untyped]
+from paramiko import Channel, channel  # type: ignore[import-untyped]
 
 from framework.logger import DTSLogger
 from framework.params import Params
 from framework.settings import SETTINGS
+from framework.testbed_model.node import Node
 
 
 class InteractiveShell(ABC):
@@ -36,13 +37,14 @@ class InteractiveShell(ABC):
     session.
     """
 
-    _interactive_session: SSHClient
+    _node: Node
     _stdin: channel.ChannelStdinFile
     _stdout: channel.ChannelFile
     _ssh_channel: Channel
     _logger: DTSLogger
     _timeout: float
     _app_params: Params
+    _privileged: bool
 
     #: Prompt to expect at the end of output when sending a command.
     #: This is often overridden by subclasses.
@@ -56,55 +58,63 @@ class InteractiveShell(ABC):
     #: 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,
+        node: Node,
         app_params: Params = Params(),
+        privileged: bool = False,
         timeout: float = SETTINGS.timeout,
+        start_on_init: bool = True,
     ) -> 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.
+            node: The node on which to run start the interactive shell.
             app_params: The command line parameters to be passed to the application on startup.
+            privileged: Enables the shell to run as superuser.
             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.
+            start_on_init: Start interactive shell automatically after object initialisation.
         """
-        self._interactive_session = interactive_session
-        self._ssh_channel = self._interactive_session.invoke_shell()
+        self._node = node
+        self._logger = node._logger
+        self._app_params = app_params
+        self._privileged = privileged
+        self._timeout = timeout
+        # Ensure path is properly formatted for the host
+        self._update_path(self._node.main_session.join_remote_path(self.path))
+
+        self._post_init()
+
+        if start_on_init:
+            self.start_application()
+
+    def _post_init(self):
+        """Overridable. Method called after the object init and before application start."""
+
+    def _setup_ssh_channel(self):
+        self._ssh_channel = self._node.main_session.interactive_session.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.settimeout(self._timeout)
         self._ssh_channel.set_combine_stderr(True)  # combines stdout and stderr streams
-        self._logger = logger
-        self._timeout = timeout
-        self._app_params = app_params
-        self._start_application(get_privileged_command)
 
-    def _start_application(self, get_privileged_command: Callable[[str], str] | None) -> None:
+    def _make_start_command(self) -> str:
+        """Makes the command that starts the interactive shell."""
+        return f"{self.path} {self._app_params or ''}"
+
+    def start_application(self) -> None:
         """Starts a new interactive application based on the path to the app.
 
         This method is often overridden by subclasses as their process for
         starting may look different.
-
-        Args:
-            get_privileged_command: A function (but could be any callable) that produces
-                the version of the command with elevated privileges.
         """
-        start_command = f"{self.path} {self._app_params}"
-        if get_privileged_command is not None:
-            start_command = get_privileged_command(start_command)
+        self._setup_ssh_channel()
+
+        start_command = self._make_start_command()
+        if self._privileged:
+            start_command = self._node.main_session._get_privileged_command(start_command)
         self.send_command(start_command)
 
     def send_command(self, command: str, prompt: str | None = None) -> str:
@@ -150,3 +160,7 @@ def close(self) -> None:
     def __del__(self) -> None:
         """Make sure the session is properly closed before deleting the object."""
         self.close()
+
+    @classmethod
+    def _update_path(cls, path: PurePath) -> None:
+        cls.path = path
diff --git a/dts/framework/remote_session/python_shell.py b/dts/framework/remote_session/python_shell.py
index ccfd3783e8..953ed100df 100644
--- a/dts/framework/remote_session/python_shell.py
+++ b/dts/framework/remote_session/python_shell.py
@@ -6,9 +6,7 @@
 Typical usage example in a TestSuite::
 
     from framework.remote_session import PythonShell
-    python_shell = self.tg_node.create_interactive_shell(
-        PythonShell, timeout=5, privileged=True
-    )
+    python_shell = PythonShell(self.tg_node, timeout=5, privileged=True)
     python_shell.send_command("print('Hello World')")
     python_shell.close()
 """
diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index ef3f23c582..39985000b9 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -7,9 +7,7 @@
 
 Typical usage example in a TestSuite::
 
-    testpmd_shell = self.sut_node.create_interactive_shell(
-            TestPmdShell, privileged=True
-        )
+    testpmd_shell = TestPmdShell(self.sut_node)
     devices = testpmd_shell.get_devices()
     for device in devices:
         print(device)
@@ -18,13 +16,14 @@
 
 import time
 from pathlib import PurePath
-from typing import Callable, ClassVar
+from typing import ClassVar
 
 from framework.exception import InteractiveCommandExecutionError
 from framework.params.testpmd import SimpleForwardingModes, TestPmdParams
+from framework.remote_session.dpdk_shell import DPDKShell
 from framework.settings import SETTINGS
-
-from .interactive_shell import InteractiveShell
+from framework.testbed_model.cpu import LogicalCoreCount, LogicalCoreList
+from framework.testbed_model.sut_node import SutNode
 
 
 class TestPmdDevice(object):
@@ -49,52 +48,48 @@ def __str__(self) -> str:
         return self.pci_address
 
 
-class TestPmdShell(InteractiveShell):
+class TestPmdShell(DPDKShell):
     """Testpmd interactive shell.
 
     The testpmd shell users should never use
     the :meth:`~.interactive_shell.InteractiveShell.send_command` method directly, but rather
     call specialized methods. If there isn't one that satisfies a need, it should be added.
-
-    Attributes:
-        number_of_ports: The number of ports which were allowed on the command-line when testpmd
-            was started.
     """
 
-    number_of_ports: int
+    _app_params: TestPmdParams
 
     #: The path to the testpmd executable.
     path: ClassVar[PurePath] = PurePath("app", "dpdk-testpmd")
 
-    #: Flag this as a DPDK app so that it's clear this is not a system app and
-    #: needs to be looked in a specific path.
-    dpdk_app: ClassVar[bool] = True
-
     #: The testpmd's prompt.
     _default_prompt: ClassVar[str] = "testpmd>"
 
     #: This forces the prompt to appear after sending a command.
     _command_extra_chars: ClassVar[str] = "\n"
 
-    def _start_application(self, get_privileged_command: Callable[[str], str] | None) -> None:
-        """Overrides :meth:`~.interactive_shell._start_application`.
-
-        Add flags for starting testpmd in interactive mode and disabling messages for link state
-        change events before starting the application. Link state is verified before starting
-        packet forwarding and the messages create unexpected newlines in the terminal which
-        complicates output collection.
-
-        Also find the number of pci addresses which were allowed on the command line when the app
-        was started.
-        """
-        assert isinstance(self._app_params, TestPmdParams)
-
-        self.number_of_ports = (
-            len(self._app_params.ports) if self._app_params.ports is not None else 0
+    def __init__(
+        self,
+        node: SutNode,
+        privileged: bool = True,
+        timeout: float = SETTINGS.timeout,
+        lcore_filter_specifier: LogicalCoreCount | LogicalCoreList = LogicalCoreCount(),
+        ascending_cores: bool = True,
+        append_prefix_timestamp: bool = True,
+        start_on_init: bool = True,
+        **app_params,
+    ) -> None:
+        """Overrides :meth:`~.dpdk_shell.DPDKShell.__init__`. Changes app_params to kwargs."""
+        super().__init__(
+            node,
+            TestPmdParams(**app_params),
+            privileged,
+            timeout,
+            lcore_filter_specifier,
+            ascending_cores,
+            append_prefix_timestamp,
+            start_on_init,
         )
 
-        super()._start_application(get_privileged_command)
-
     def start(self, verify: bool = True) -> None:
         """Start packet forwarding with the current configuration.
 
@@ -114,7 +109,8 @@ def start(self, verify: bool = True) -> None:
                 self._logger.debug(f"Failed to start packet forwarding: \n{start_cmd_output}")
                 raise InteractiveCommandExecutionError("Testpmd failed to start packet forwarding.")
 
-            for port_id in range(self.number_of_ports):
+            number_of_ports = len(self._app_params.ports or [])
+            for port_id in range(number_of_ports):
                 if not self.wait_link_status_up(port_id):
                     raise InteractiveCommandExecutionError(
                         "Not all ports came up after starting packet forwarding in testpmd."
diff --git a/dts/framework/testbed_model/node.py b/dts/framework/testbed_model/node.py
index 6af4f25a3c..88395faabe 100644
--- a/dts/framework/testbed_model/node.py
+++ b/dts/framework/testbed_model/node.py
@@ -15,7 +15,7 @@
 
 from abc import ABC
 from ipaddress import IPv4Interface, IPv6Interface
-from typing import Any, Callable, Type, Union
+from typing import Any, Callable, Union
 
 from framework.config import (
     OS,
@@ -25,7 +25,6 @@
 )
 from framework.exception import ConfigurationError
 from framework.logger import DTSLogger, get_dts_logger
-from framework.params import Params
 from framework.settings import SETTINGS
 
 from .cpu import (
@@ -36,7 +35,7 @@
     lcore_filter,
 )
 from .linux_session import LinuxSession
-from .os_session import InteractiveShellType, OSSession
+from .os_session import OSSession
 from .port import Port
 from .virtual_device import VirtualDevice
 
@@ -196,37 +195,6 @@ 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_params: Params = Params(),
-    ) -> InteractiveShellType:
-        """Factory for interactive session handlers.
-
-        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:
-            An instance of the desired interactive application shell.
-        """
-        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,
-            timeout,
-            privileged,
-            app_params,
-        )
-
     def filter_lcores(
         self,
         filter_specifier: LogicalCoreCount | LogicalCoreList,
diff --git a/dts/framework/testbed_model/os_session.py b/dts/framework/testbed_model/os_session.py
index e5f5fcbe0e..e7e6c9d670 100644
--- a/dts/framework/testbed_model/os_session.py
+++ b/dts/framework/testbed_model/os_session.py
@@ -26,18 +26,16 @@
 from collections.abc import Iterable
 from ipaddress import IPv4Interface, IPv6Interface
 from pathlib import PurePath
-from typing import Type, TypeVar, Union
+from typing import Union
 
 from framework.config import Architecture, NodeConfiguration, NodeInfo
 from framework.logger import DTSLogger
-from framework.params import Params
 from framework.remote_session import (
     InteractiveRemoteSession,
     RemoteSession,
     create_interactive_session,
     create_remote_session,
 )
-from framework.remote_session.interactive_shell import InteractiveShell
 from framework.remote_session.remote_session import CommandResult
 from framework.settings import SETTINGS
 from framework.utils import MesonArgs
@@ -45,8 +43,6 @@
 from .cpu import LogicalCore
 from .port import Port
 
-InteractiveShellType = TypeVar("InteractiveShellType", bound=InteractiveShell)
-
 
 class OSSession(ABC):
     """OS-unaware to OS-aware translation API definition.
@@ -131,36 +127,6 @@ def send_command(
 
         return self.remote_session.send_command(command, timeout, verify, env)
 
-    def create_interactive_shell(
-        self,
-        shell_cls: Type[InteractiveShellType],
-        timeout: float,
-        privileged: bool,
-        app_args: Params,
-    ) -> InteractiveShellType:
-        """Factory for interactive session handlers.
-
-        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:
-            An instance of the desired interactive application shell.
-        """
-        return shell_cls(
-            self.interactive_session.session,
-            self._logger,
-            self._get_privileged_command if privileged else None,
-            app_args,
-            timeout,
-        )
-
     @staticmethod
     @abstractmethod
     def _get_privileged_command(command: str) -> str:
diff --git a/dts/framework/testbed_model/sut_node.py b/dts/framework/testbed_model/sut_node.py
index 83ad06ae2d..727170b7fc 100644
--- a/dts/framework/testbed_model/sut_node.py
+++ b/dts/framework/testbed_model/sut_node.py
@@ -16,7 +16,6 @@
 import tarfile
 import time
 from pathlib import PurePath
-from typing import Type
 
 from framework.config import (
     BuildTargetConfiguration,
@@ -24,17 +23,13 @@
     NodeInfo,
     SutNodeConfiguration,
 )
-from framework.params import Params, Switch
 from framework.params.eal import EalParams
 from framework.remote_session.remote_session import CommandResult
 from framework.settings import SETTINGS
 from framework.utils import MesonArgs
 
-from .cpu import LogicalCoreCount, LogicalCoreList
 from .node import Node
-from .os_session import InteractiveShellType, OSSession
-from .port import Port
-from .virtual_device import VirtualDevice
+from .os_session import OSSession
 
 
 class SutNode(Node):
@@ -289,68 +284,6 @@ def kill_cleanup_dpdk_apps(self) -> None:
             self._dpdk_kill_session = self.create_session("dpdk_kill")
         self._dpdk_prefix_list = []
 
-    def create_eal_parameters(
-        self,
-        lcore_filter_specifier: LogicalCoreCount | LogicalCoreList = LogicalCoreCount(),
-        ascending_cores: bool = True,
-        prefix: str = "dpdk",
-        append_prefix_timestamp: bool = True,
-        no_pci: Switch = None,
-        vdevs: list[VirtualDevice] | None = None,
-        ports: list[Port] | None = None,
-        other_eal_param: str = "",
-    ) -> EalParams:
-        """Compose the EAL parameters.
-
-        Process the list of cores and the DPDK prefix and pass that along with
-        the rest of the arguments.
-
-        Args:
-            lcore_filter_specifier: A number of lcores/cores/sockets to use
-                or a list of lcore ids to use.
-                The default will select one lcore for each of two cores
-                on one socket, in ascending order of core ids.
-            ascending_cores: Sort cores in ascending order (lowest to highest IDs).
-                If :data:`False`, sort in descending order.
-            prefix: Set the file prefix string with which to start DPDK, e.g.: ``prefix='vf'``.
-            append_prefix_timestamp: If :data:`True`, will append a timestamp to DPDK file prefix.
-            no_pci: Switch to disable PCI bus e.g.: ``no_pci=True``.
-            vdevs: Virtual devices, e.g.::
-
-                vdevs=[
-                    VirtualDevice('net_ring0'),
-                    VirtualDevice('net_ring1')
-                ]
-            ports: The list of ports to allow. If :data:`None`, all ports listed in `self.ports`
-                will be allowed.
-            other_eal_param: user defined DPDK EAL parameters, e.g.:
-                ``other_eal_param='--single-file-segments'``.
-
-        Returns:
-            An EAL param string, such as
-            ``-c 0xf -a 0000:88:00.0 --file-prefix=dpdk_1112_20190809143420``.
-        """
-        lcore_list = LogicalCoreList(self.filter_lcores(lcore_filter_specifier, ascending_cores))
-
-        if append_prefix_timestamp:
-            prefix = f"{prefix}_{self._dpdk_timestamp}"
-        prefix = self.main_session.get_dpdk_file_prefix(prefix)
-        if prefix:
-            self._dpdk_prefix_list.append(prefix)
-
-        if ports is None:
-            ports = self.ports
-
-        return EalParams(
-            lcore_list=lcore_list,
-            memory_channels=self.config.memory_channels,
-            prefix=prefix,
-            no_pci=no_pci,
-            vdevs=vdevs,
-            ports=ports,
-            other_eal_param=Params.from_str(other_eal_param),
-        )
-
     def run_dpdk_app(
         self, app_path: PurePath, eal_params: EalParams, timeout: float = 30
     ) -> CommandResult:
@@ -379,49 +312,6 @@ def configure_ipv4_forwarding(self, enable: bool) -> None:
         """
         self.main_session.configure_ipv4_forwarding(enable)
 
-    def create_interactive_shell(
-        self,
-        shell_cls: Type[InteractiveShellType],
-        timeout: float = SETTINGS.timeout,
-        privileged: bool = False,
-        app_params: Params = Params(),
-        eal_params: EalParams | None = None,
-    ) -> InteractiveShellType:
-        """Extend the factory for interactive session handlers.
-
-        The extensions are SUT node specific:
-
-            * The default for `eal_parameters`,
-            * The interactive shell path `shell_cls.path` is prepended with path to the remote
-              DPDK build directory for DPDK apps.
-
-        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_params: The parameters to be passed to the application.
-            eal_params: List of EAL parameters to use to launch the app. If this
-                isn't provided or an empty string is passed, it will default to calling
-                :meth:`create_eal_parameters`.
-
-        Returns:
-            An instance of the desired interactive application shell.
-        """
-        # We need to append the build directory and add EAL parameters for DPDK apps
-        if shell_cls.dpdk_app:
-            if eal_params is None:
-                eal_params = self.create_eal_parameters()
-            eal_params.append_str(str(app_params))
-            app_params = eal_params
-
-            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, app_params)
-
     def bind_ports_to_driver(self, for_dpdk: bool = True) -> None:
         """Bind all ports on the SUT to a driver.
 
diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
index 7bc1c2cc08..bf58ad1c5e 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -217,9 +217,7 @@ def __init__(self, tg_node: Node, config: ScapyTrafficGeneratorConfig):
             self._tg_node.config.os == OS.linux
         ), "Linux is the only supported OS for scapy traffic generation"
 
-        self.session = self._tg_node.create_interactive_shell(
-            PythonShell, timeout=5, privileged=True
-        )
+        self.session = PythonShell(self._tg_node, timeout=5, privileged=True)
 
         # import libs in remote python console
         for import_statement in SCAPY_RPC_SERVER_IMPORTS:
diff --git a/dts/tests/TestSuite_hello_world.py b/dts/tests/TestSuite_hello_world.py
index 0d6995f260..d958f99030 100644
--- a/dts/tests/TestSuite_hello_world.py
+++ b/dts/tests/TestSuite_hello_world.py
@@ -7,6 +7,7 @@
 No other EAL parameters apart from cores are used.
 """
 
+from framework.remote_session.dpdk_shell import compute_eal_params
 from framework.test_suite import TestSuite
 from framework.testbed_model.cpu import (
     LogicalCoreCount,
@@ -38,7 +39,7 @@ def test_hello_world_single_core(self) -> None:
         # get the first usable core
         lcore_amount = LogicalCoreCount(1, 1, 1)
         lcores = LogicalCoreCountFilter(self.sut_node.lcores, lcore_amount).filter()
-        eal_para = self.sut_node.create_eal_parameters(lcore_filter_specifier=lcore_amount)
+        eal_para = compute_eal_params(self.sut_node, lcore_filter_specifier=lcore_amount)
         result = self.sut_node.run_dpdk_app(self.app_helloworld_path, eal_para)
         self.verify(
             f"hello from core {int(lcores[0])}" in result.stdout,
@@ -55,8 +56,8 @@ def test_hello_world_all_cores(self) -> None:
             "hello from core <core_id>"
         """
         # get the maximum logical core number
-        eal_para = self.sut_node.create_eal_parameters(
-            lcore_filter_specifier=LogicalCoreList(self.sut_node.lcores)
+        eal_para = compute_eal_params(
+            self.sut_node, lcore_filter_specifier=LogicalCoreList(self.sut_node.lcores)
         )
         result = self.sut_node.run_dpdk_app(self.app_helloworld_path, eal_para, 50)
         for lcore in self.sut_node.lcores:
diff --git a/dts/tests/TestSuite_pmd_buffer_scatter.py b/dts/tests/TestSuite_pmd_buffer_scatter.py
index 6d206c1a40..43cf5c61eb 100644
--- a/dts/tests/TestSuite_pmd_buffer_scatter.py
+++ b/dts/tests/TestSuite_pmd_buffer_scatter.py
@@ -16,14 +16,13 @@
 """
 
 import struct
-from dataclasses import asdict
 
 from scapy.layers.inet import IP  # type: ignore[import-untyped]
 from scapy.layers.l2 import Ether  # type: ignore[import-untyped]
 from scapy.packet import Raw  # type: ignore[import-untyped]
 from scapy.utils import hexstr  # type: ignore[import-untyped]
 
-from framework.params.testpmd import SimpleForwardingModes, TestPmdParams
+from framework.params.testpmd import SimpleForwardingModes
 from framework.remote_session.testpmd_shell import TestPmdShell
 from framework.test_suite import TestSuite
 
@@ -103,17 +102,13 @@ def pmd_scatter(self, mbsize: int) -> None:
         Test:
             Start testpmd and run functional test with preset mbsize.
         """
-        testpmd = self.sut_node.create_interactive_shell(
-            TestPmdShell,
-            app_params=TestPmdParams(
-                forward_mode=SimpleForwardingModes.mac,
-                mbcache=200,
-                mbuf_size=[mbsize],
-                max_pkt_len=9000,
-                tx_offloads=0x00008000,
-                **asdict(self.sut_node.create_eal_parameters()),
-            ),
-            privileged=True,
+        testpmd = TestPmdShell(
+            self.sut_node,
+            forward_mode=SimpleForwardingModes.mac,
+            mbcache=200,
+            mbuf_size=[mbsize],
+            max_pkt_len=9000,
+            tx_offloads=0x00008000,
         )
         testpmd.start()
 
diff --git a/dts/tests/TestSuite_smoke_tests.py b/dts/tests/TestSuite_smoke_tests.py
index ca678f662d..eca27acfd8 100644
--- a/dts/tests/TestSuite_smoke_tests.py
+++ b/dts/tests/TestSuite_smoke_tests.py
@@ -99,7 +99,7 @@ def test_devices_listed_in_testpmd(self) -> None:
         Test:
             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)
+        testpmd_driver = TestPmdShell(self.sut_node)
         dev_list = [str(x) for x in testpmd_driver.get_devices()]
         for nic in self.nics_in_node:
             self.verify(
-- 
2.34.1