[v9,1/7] dts: add startup verification and forwarding modes to testpmd shell

diff --git a/dts/framework/exception.py b/dts/framework/exception.py
index 658eee2c38..cce1e0231a 100644
--- a/dts/framework/exception.py
+++ b/dts/framework/exception.py
@@ -146,6 +146,13 @@  def __str__(self) -> str:
         return f"Command {self.command} returned a non-zero exit code: {self._command_return_code}"
 
 
+class InteractiveCommandExecutionError(DTSError):
+    """An unsuccessful execution of a remote command in an interactive environment."""
+
+    #:
+    severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR
+
+
 class RemoteDirectoryExistsError(DTSError):
     """A directory that exists on a remote node."""
 
diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index 0184cc2e71..3a66907394 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -15,9 +15,15 @@ 
     testpmd_shell.close()
 """
 
+import time
+from enum import auto
 from pathlib import PurePath
 from typing import Callable, ClassVar
 
+from framework.exception import InteractiveCommandExecutionError
+from framework.settings import SETTINGS
+from framework.utils import StrEnum
+
 from .interactive_shell import InteractiveShell
 
 
@@ -43,14 +49,51 @@  def __str__(self) -> str:
         return self.pci_address
 
 
+class TestPmdForwardingModes(StrEnum):
+    r"""The supported packet forwarding modes for :class:`~TestPmdShell`\s."""
+
+    #:
+    io = auto()
+    #:
+    mac = auto()
+    #:
+    macswap = auto()
+    #:
+    flowgen = auto()
+    #:
+    rxonly = auto()
+    #:
+    txonly = auto()
+    #:
+    csum = auto()
+    #:
+    icmpecho = auto()
+    #:
+    ieee1588 = auto()
+    #:
+    noisy = auto()
+    #:
+    fivetswap = "5tswap"
+    #:
+    shared_rxq = "shared-rxq"
+    #:
+    recycle_mbufs = auto()
+
+
 class TestPmdShell(InteractiveShell):
     """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
+
     #: The path to the testpmd executable.
     path: ClassVar[PurePath] = PurePath("app", "dpdk-testpmd")
 
@@ -65,9 +108,66 @@  class TestPmdShell(InteractiveShell):
     _command_extra_chars: ClassVar[str] = "\n"
 
     def _start_application(self, get_privileged_command: Callable[[str], str] | None) -> None:
-        self._app_args += " -- -i"
+        """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.
+        """
+        self._app_args += " -- -i --mask-event intr_lsc"
+        self.number_of_ports = self._app_args.count("-a ")
         super()._start_application(get_privileged_command)
 
+    def start(self, verify: bool = True) -> None:
+        """Start packet forwarding with the current configuration.
+
+        Args:
+            verify: If :data:`True` , a second start command will be sent in an attempt to verify
+                packet forwarding started as expected.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and forwarding fails to
+                start or ports fail to come up.
+        """
+        self.send_command("start")
+        if verify:
+            # If forwarding was already started, sending "start" again should tell us
+            start_cmd_output = self.send_command("start")
+            if "Packet forwarding already started" not in start_cmd_output:
+                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):
+                if not self.wait_link_status_up(port_id):
+                    raise InteractiveCommandExecutionError(
+                        "Not all ports came up after starting packet forwarding in testpmd."
+                    )
+
+    def stop(self, verify: bool = True) -> None:
+        """Stop packet forwarding.
+
+        Args:
+            verify: If :data:`True` , the output of the stop command is scanned to verify that
+                forwarding was stopped successfully or not started. If neither is found, it is
+                considered an error.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the command to stop
+                forwarding results in an error.
+        """
+        stop_cmd_output = self.send_command("stop")
+        if verify:
+            if (
+                "Done." not in stop_cmd_output
+                and "Packet forwarding not started" not in stop_cmd_output
+            ):
+                self._logger.debug(f"Failed to stop packet forwarding: \n{stop_cmd_output}")
+                raise InteractiveCommandExecutionError("Testpmd failed to stop packet forwarding.")
+
     def get_devices(self) -> list[TestPmdDevice]:
         """Get a list of device names that are known to testpmd.
 
@@ -82,3 +182,50 @@  def get_devices(self) -> list[TestPmdDevice]:
             if "device name:" in line.lower():
                 dev_list.append(TestPmdDevice(line))
         return dev_list
+
+    def wait_link_status_up(self, port_id: int, timeout=SETTINGS.timeout) -> bool:
+        """Wait until the link status on the given port is "up".
+
+        Arguments:
+            port_id: Port to check the link status on.
+            timeout: Time to wait for the link to come up. The default value for this
+                argument may be modified using the :option:`--timeout` command-line argument
+                or the :envvar:`DTS_TIMEOUT` environment variable.
+
+        Returns:
+            Whether the link came up in time or not.
+        """
+        time_to_stop = time.time() + timeout
+        port_info: str = ""
+        while time.time() < time_to_stop:
+            port_info = self.send_command(f"show port info {port_id}")
+            if "Link status: up" in port_info:
+                break
+            time.sleep(0.5)
+        else:
+            self._logger.error(f"The link for port {port_id} did not come up in the given timeout.")
+        return "Link status: up" in port_info
+
+    def set_forward_mode(self, mode: TestPmdForwardingModes, verify: bool = True):
+        """Set packet forwarding mode.
+
+        Args:
+            mode: The forwarding mode to use.
+            verify: If :data:`True` the output of the command will be scanned in an attempt to
+                verify that the forwarding mode was set to `mode` properly.
+
+        Raises:
+            InteractiveCommandExecutionError: If `verify` is :data:`True` and the forwarding mode
+                fails to update.
+        """
+        set_fwd_output = self.send_command(f"set fwd {mode.value}")
+        if f"Set {mode.value} packet forwarding mode" not in set_fwd_output:
+            self._logger.debug(f"Failed to set fwd mode to {mode.value}:\n{set_fwd_output}")
+            raise InteractiveCommandExecutionError(
+                f"Test pmd failed to set fwd mode to {mode.value}"
+            )
+
+    def close(self) -> None:
+        """Overrides :meth:`~.interactive_shell.close`."""
+        self.send_command("quit", "")
+        return super().close()