1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2023 University of New Hampshire 3# Copyright(c) 2023 PANTHEON.tech s.r.o. 4 5"""Testpmd interactive shell. 6 7Typical usage example in a TestSuite:: 8 9 testpmd_shell = self.sut_node.create_interactive_shell( 10 TestPmdShell, privileged=True 11 ) 12 devices = testpmd_shell.get_devices() 13 for device in devices: 14 print(device) 15 testpmd_shell.close() 16""" 17 18import time 19from enum import auto 20from pathlib import PurePath 21from typing import Callable, ClassVar 22 23from framework.exception import InteractiveCommandExecutionError 24from framework.settings import SETTINGS 25from framework.utils import StrEnum 26 27from .interactive_shell import InteractiveShell 28 29 30class TestPmdDevice(object): 31 """The data of a device that testpmd can recognize. 32 33 Attributes: 34 pci_address: The PCI address of the device. 35 """ 36 37 pci_address: str 38 39 def __init__(self, pci_address_line: str): 40 """Initialize the device from the testpmd output line string. 41 42 Args: 43 pci_address_line: A line of testpmd output that contains a device. 44 """ 45 self.pci_address = pci_address_line.strip().split(": ")[1].strip() 46 47 def __str__(self) -> str: 48 """The PCI address captures what the device is.""" 49 return self.pci_address 50 51 52class TestPmdForwardingModes(StrEnum): 53 r"""The supported packet forwarding modes for :class:`~TestPmdShell`\s.""" 54 55 #: 56 io = auto() 57 #: 58 mac = auto() 59 #: 60 macswap = auto() 61 #: 62 flowgen = auto() 63 #: 64 rxonly = auto() 65 #: 66 txonly = auto() 67 #: 68 csum = auto() 69 #: 70 icmpecho = auto() 71 #: 72 ieee1588 = auto() 73 #: 74 noisy = auto() 75 #: 76 fivetswap = "5tswap" 77 #: 78 shared_rxq = "shared-rxq" 79 #: 80 recycle_mbufs = auto() 81 82 83class TestPmdShell(InteractiveShell): 84 """Testpmd interactive shell. 85 86 The testpmd shell users should never use 87 the :meth:`~.interactive_shell.InteractiveShell.send_command` method directly, but rather 88 call specialized methods. If there isn't one that satisfies a need, it should be added. 89 90 Attributes: 91 number_of_ports: The number of ports which were allowed on the command-line when testpmd 92 was started. 93 """ 94 95 number_of_ports: int 96 97 #: The path to the testpmd executable. 98 path: ClassVar[PurePath] = PurePath("app", "dpdk-testpmd") 99 100 #: Flag this as a DPDK app so that it's clear this is not a system app and 101 #: needs to be looked in a specific path. 102 dpdk_app: ClassVar[bool] = True 103 104 #: The testpmd's prompt. 105 _default_prompt: ClassVar[str] = "testpmd>" 106 107 #: This forces the prompt to appear after sending a command. 108 _command_extra_chars: ClassVar[str] = "\n" 109 110 def _start_application(self, get_privileged_command: Callable[[str], str] | None) -> None: 111 """Overrides :meth:`~.interactive_shell._start_application`. 112 113 Add flags for starting testpmd in interactive mode and disabling messages for link state 114 change events before starting the application. Link state is verified before starting 115 packet forwarding and the messages create unexpected newlines in the terminal which 116 complicates output collection. 117 118 Also find the number of pci addresses which were allowed on the command line when the app 119 was started. 120 """ 121 self._app_args += " -i --mask-event intr_lsc" 122 self.number_of_ports = self._app_args.count("-a ") 123 super()._start_application(get_privileged_command) 124 125 def start(self, verify: bool = True) -> None: 126 """Start packet forwarding with the current configuration. 127 128 Args: 129 verify: If :data:`True` , a second start command will be sent in an attempt to verify 130 packet forwarding started as expected. 131 132 Raises: 133 InteractiveCommandExecutionError: If `verify` is :data:`True` and forwarding fails to 134 start or ports fail to come up. 135 """ 136 self.send_command("start") 137 if verify: 138 # If forwarding was already started, sending "start" again should tell us 139 start_cmd_output = self.send_command("start") 140 if "Packet forwarding already started" not in start_cmd_output: 141 self._logger.debug(f"Failed to start packet forwarding: \n{start_cmd_output}") 142 raise InteractiveCommandExecutionError("Testpmd failed to start packet forwarding.") 143 144 for port_id in range(self.number_of_ports): 145 if not self.wait_link_status_up(port_id): 146 raise InteractiveCommandExecutionError( 147 "Not all ports came up after starting packet forwarding in testpmd." 148 ) 149 150 def stop(self, verify: bool = True) -> None: 151 """Stop packet forwarding. 152 153 Args: 154 verify: If :data:`True` , the output of the stop command is scanned to verify that 155 forwarding was stopped successfully or not started. If neither is found, it is 156 considered an error. 157 158 Raises: 159 InteractiveCommandExecutionError: If `verify` is :data:`True` and the command to stop 160 forwarding results in an error. 161 """ 162 stop_cmd_output = self.send_command("stop") 163 if verify: 164 if ( 165 "Done." not in stop_cmd_output 166 and "Packet forwarding not started" not in stop_cmd_output 167 ): 168 self._logger.debug(f"Failed to stop packet forwarding: \n{stop_cmd_output}") 169 raise InteractiveCommandExecutionError("Testpmd failed to stop packet forwarding.") 170 171 def get_devices(self) -> list[TestPmdDevice]: 172 """Get a list of device names that are known to testpmd. 173 174 Uses the device info listed in testpmd and then parses the output. 175 176 Returns: 177 A list of devices. 178 """ 179 dev_info: str = self.send_command("show device info all") 180 dev_list: list[TestPmdDevice] = [] 181 for line in dev_info.split("\n"): 182 if "device name:" in line.lower(): 183 dev_list.append(TestPmdDevice(line)) 184 return dev_list 185 186 def wait_link_status_up(self, port_id: int, timeout=SETTINGS.timeout) -> bool: 187 """Wait until the link status on the given port is "up". 188 189 Arguments: 190 port_id: Port to check the link status on. 191 timeout: Time to wait for the link to come up. The default value for this 192 argument may be modified using the :option:`--timeout` command-line argument 193 or the :envvar:`DTS_TIMEOUT` environment variable. 194 195 Returns: 196 Whether the link came up in time or not. 197 """ 198 time_to_stop = time.time() + timeout 199 port_info: str = "" 200 while time.time() < time_to_stop: 201 port_info = self.send_command(f"show port info {port_id}") 202 if "Link status: up" in port_info: 203 break 204 time.sleep(0.5) 205 else: 206 self._logger.error(f"The link for port {port_id} did not come up in the given timeout.") 207 return "Link status: up" in port_info 208 209 def set_forward_mode(self, mode: TestPmdForwardingModes, verify: bool = True): 210 """Set packet forwarding mode. 211 212 Args: 213 mode: The forwarding mode to use. 214 verify: If :data:`True` the output of the command will be scanned in an attempt to 215 verify that the forwarding mode was set to `mode` properly. 216 217 Raises: 218 InteractiveCommandExecutionError: If `verify` is :data:`True` and the forwarding mode 219 fails to update. 220 """ 221 set_fwd_output = self.send_command(f"set fwd {mode.value}") 222 if f"Set {mode.value} packet forwarding mode" not in set_fwd_output: 223 self._logger.debug(f"Failed to set fwd mode to {mode.value}:\n{set_fwd_output}") 224 raise InteractiveCommandExecutionError( 225 f"Test pmd failed to set fwd mode to {mode.value}" 226 ) 227 228 def close(self) -> None: 229 """Overrides :meth:`~.interactive_shell.close`.""" 230 self.send_command("quit", "") 231 return super().close() 232