Add a new TestPmdPortStats data structure to represent the output returned by `show port stats`, which is implemented as part of TestPmdShell.
Bugzilla ID: 1407 Signed-off-by: Luca Vizzarro <luca.vizza...@arm.com> Reviewed-by: Paul Szczepanek <paul.szczepa...@arm.com> --- dts/framework/remote_session/testpmd_shell.py | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py index 3cf123ff57..baf47d1a32 100644 --- a/dts/framework/remote_session/testpmd_shell.py +++ b/dts/framework/remote_session/testpmd_shell.py @@ -519,6 +519,42 @@ class TestPmdPort(TextParser): ) +@dataclass +class TestPmdPortStats(TextParser): + """Port statistics.""" + + #: + port_id: int = field(metadata=parser.to_int(parser.regex(r"NIC statistics for port (\d+)"))) + + #: + rx_packets: int = field(metadata=parser.to_int(parser.regex(r"RX-packets:\s+(\d+)"))) + #: + rx_missed: int = field(metadata=parser.to_int(parser.regex(r"RX-missed:\s+(\d+)"))) + #: + rx_bytes: int = field(metadata=parser.to_int(parser.regex(r"RX-bytes:\s+(\d+)"))) + #: + rx_errors: int = field(metadata=parser.to_int(parser.regex(r"RX-errors:\s+(\d+)"))) + #: + rx_nombuf: int = field(metadata=parser.to_int(parser.regex(r"RX-nombuf:\s+(\d+)"))) + + #: + tx_packets: int = field(metadata=parser.to_int(parser.regex(r"TX-packets:\s+(\d+)"))) + #: + tx_errors: int = field(metadata=parser.to_int(parser.regex(r"TX-errors:\s+(\d+)"))) + #: + tx_bytes: int = field(metadata=parser.to_int(parser.regex(r"TX-bytes:\s+(\d+)"))) + + #: + rx_pps: int = field(metadata=parser.to_int(parser.regex(r"Rx-pps:\s+(\d+)"))) + #: + rx_bps: int = field(metadata=parser.to_int(parser.regex(r"Rx-bps:\s+(\d+)"))) + + #: + tx_pps: int = field(metadata=parser.to_int(parser.regex(r"Tx-pps:\s+(\d+)"))) + #: + tx_bps: int = field(metadata=parser.to_int(parser.regex(r"Tx-bps:\s+(\d+)"))) + + class TestPmdShell(InteractiveShell): """Testpmd interactive shell. @@ -696,6 +732,28 @@ def show_port_info(self, port_id: int) -> TestPmdPort: return TestPmdPort.parse(output) + def show_port_stats_all(self) -> list[TestPmdPortStats]: + """Returns the statistics of all the ports.""" + output = self.send_command("show port stats all") + + iter = re.finditer(r"(^ #*.+#*$[^#]+)^ #*$", output, re.MULTILINE) + return [TestPmdPortStats.parse(block.group(1)) for block in iter] + + def show_port_stats(self, port_id: int) -> TestPmdPortStats: + """Returns the given port statistics. + + Args: + port_id: The port ID to gather information for. + + Raises: + InteractiveCommandExecutionError: If `port_id` is invalid. + """ + output = self.send_command(f"show port stats {port_id}") + if output.startswith("Invalid port"): + raise InteractiveCommandExecutionError("invalid port given") + + return TestPmdPortStats.parse(output) + def close(self) -> None: """Overrides :meth:`~.interactive_shell.close`.""" self.send_command("quit", "") -- 2.34.1