Methods currently exist for querying the statistics of a port in
testpmd, but there weren't methods added for clearing the current
statistics on a port. This patch adds methods that allow you to clear
the statistics of a single port or all ports to account for situations
where the user only wants the port statistics after a certain point and
does not care about any existing prior values.
This patch also modifies the show_port_stats_all method to allow for it
to return the raw testpmd output gathered from sending the command as
well as the parsed output. This allows users to access and examine
additional information provided by the raw output if needed.
Signed-off-by: Jeremy Spewock <jspew...@iol.unh.edu>
Signed-off-by: Dean Marx <dm...@iol.unh.edu>
---
dts/framework/remote_session/testpmd_shell.py | 57 +++++++++++++++----
1 file changed, 45 insertions(+), 12 deletions(-)
diff --git a/dts/framework/remote_session/testpmd_shell.py
b/dts/framework/remote_session/testpmd_shell.py
index 672ecd970f..b8e37434ab 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -22,15 +22,7 @@
from enum import Flag, auto
from os import environ
from pathlib import PurePath
-from typing import (
- TYPE_CHECKING,
- Any,
- ClassVar,
- Concatenate,
- Literal,
- ParamSpec,
- TypeAlias,
-)
+from typing import TYPE_CHECKING, Any, ClassVar, Concatenate, Literal,
ParamSpec, Tuple, TypeAlias
from framework.context import get_ctx
from framework.testbed_model.topology import TopologyType
@@ -1836,11 +1828,13 @@ def set_multicast_mac_addr(
f"Failed to {mcast_cmd} {multi_addr} on port {port_id}
\n{output}"
)
- def show_port_stats_all(self) -> list[TestPmdPortStats]:
+ def show_port_stats_all(self) -> Tuple[list[TestPmdPortStats], str]:
"""Returns the statistics of all the ports.
Returns:
- list[TestPmdPortStats]: A list containing all the ports stats as
`TestPmdPortStats`.
+ Tuple[str, list[TestPmdPortStats]]: A tuple where the first
element is the stats of all
+ ports as `TestPmdPortStats` and second is the raw testpmd output
that was collected
+ from the sent command.
"""
output = self.send_command("show port stats all")
@@ -1855,7 +1849,7 @@ def show_port_stats_all(self) -> list[TestPmdPortStats]:
# #################################################
#
iter = re.finditer(r"(^ #*.+#*$[^#]+)^ #*\r$", output, re.MULTILINE)
- return [TestPmdPortStats.parse(block.group(1)) for block in iter]
+ return ([TestPmdPortStats.parse(block.group(1)) for block in iter],
output)
def show_port_stats(self, port_id: int) -> TestPmdPortStats:
"""Returns the given port statistics.
@@ -2312,6 +2306,45 @@ def rx_vxlan(self, vxlan_id: int, port_id: int, enable:
bool, verify: bool = Tru
self._logger.debug(f"Failed to set VXLAN:\n{vxlan_output}")
raise InteractiveCommandExecutionError(f"Failed to set
VXLAN:\n{vxlan_output}")
+ def clear_port_stats(self, port_id: int, verify: bool = True) -> None:
+ """Clear statistics of a given port.
+
+ Args:
+ port_id: ID of the port to clear the statistics on.
+ verify: If :data:`True` the output of the command will be scanned
to verify that it was
+ successful, otherwise failures will be ignored. Defaults to
:data:`True`.
+
+ Raises:
+ InteractiveCommandExecutionError: If `verify` is :data:`True` and
testpmd fails to
+ clear the statistics of the given port.
+ """
+ clear_output = self.send_command(f"clear port stats {port_id}")
+ if verify and f"NIC statistics for port {port_id} cleared" not in
clear_output:
+ raise InteractiveCommandExecutionError(
+ f"Test pmd failed to set clear forwarding stats on port
{port_id}"
+ )
+
+ def clear_port_stats_all(self, verify: bool = True) -> None:
+ """Clear the statistics of all ports that testpmd is aware of.
+
+ Args:
+ verify: If :data:`True` the output of the command will be scanned
to verify that all
+ ports had their statistics cleared, otherwise failures will be
ignored. Defaults to
+ :data:`True`.
+
+ Raises:
+ InteractiveCommandExecutionError: If `verify` is :data:`True` and
testpmd fails to
+ clear the statistics of any of its ports.
+ """
+ clear_output = self.send_command("clear port stats all")
+ if verify:
+ if type(self._app_params.port_numa_config) is list:
+ for port_id in range(len(self._app_params.port_numa_config)):
+ if f"NIC statistics for port {port_id} cleared" not in
clear_output:
+ raise InteractiveCommandExecutionError(
+ f"Test pmd failed to set clear forwarding stats on
port {port_id}"
+ )
+
def _close(self) -> None:
"""Overrides :meth:`~.interactive_shell.close`."""
self.stop()
--
2.47.0
