1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2023 University of New Hampshire
# Copyright(c) 2023 PANTHEON.tech s.r.o.
"""Testpmd interactive shell.
Typical usage example in a TestSuite::
testpmd_shell = self.sut_node.create_interactive_shell(
TestPmdShell, privileged=True
)
devices = testpmd_shell.get_devices()
for device in devices:
print(device)
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
class TestPmdDevice(object):
"""The data of a device that testpmd can recognize.
Attributes:
pci_address: The PCI address of the device.
"""
pci_address: str
def __init__(self, pci_address_line: str):
"""Initialize the device from the testpmd output line string.
Args:
pci_address_line: A line of testpmd output that contains a device.
"""
self.pci_address = pci_address_line.strip().split(": ")[1].strip()
def __str__(self) -> str:
"""The PCI address captures what the device is."""
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")
#: 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.
"""
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.
Uses the device info listed in testpmd and then parses the output.
Returns:
A list of devices.
"""
dev_info: str = self.send_command("show device info all")
dev_list: list[TestPmdDevice] = []
for line in dev_info.split("\n"):
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()
|
