When using port information multiple times in a testpmd shell instance
lifespan, it's desirable to not get the information each time, so
caching is added. In case the information changes, there's a way to
force the update.
Signed-off-by: Juraj Linkeš <juraj.lin...@pantheon.tech>
---
dts/framework/remote_session/testpmd_shell.py | 30 +++++++++++++++++--
1 file changed, 28 insertions(+), 2 deletions(-)
diff --git a/dts/framework/remote_session/testpmd_shell.py
b/dts/framework/remote_session/testpmd_shell.py
index b4ad253020..f0bcc918e5 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -654,6 +654,7 @@ class TestPmdShell(DPDKShell):
"""
_app_params: TestPmdParams
+ _ports: list[TestPmdPort] | None
#: The path to the testpmd executable.
path: ClassVar[PurePath] = PurePath("app", "dpdk-testpmd")
@@ -686,6 +687,21 @@ def __init__(
TestPmdParams(**app_params),
name,
)
+ self._ports = None
+
+ @property
+ def ports(self) -> list[TestPmdPort]:
+ """The ports of the instance.
+
+ This caches the ports returned by :meth:`show_port_info_all`.
+ To force an update of port information, execute
:meth:`show_port_info_all` or
+ :meth:`show_port_info`.
+
+ Returns: The list of known testpmd ports.
+ """
+ if self._ports is None:
+ return self.show_port_info_all()
+ return self._ports
def start(self, verify: bool = True) -> None:
"""Start packet forwarding with the current configuration.
@@ -872,7 +888,8 @@ def show_port_info_all(self) -> list[TestPmdPort]:
# executed on a pseudo-terminal created by paramiko on the remote
node, lines end with CRLF.
# Therefore we also need to take the carriage return into account.
iter = re.finditer(r"\*{21}.*?[\r\n]{4}", output + "\r\n", re.S)
- return [TestPmdPort.parse(block.group(0)) for block in iter]
+ self._ports = [TestPmdPort.parse(block.group(0)) for block in iter]
+ return self._ports
def show_port_info(self, port_id: int) -> TestPmdPort:
"""Returns the given port information.
@@ -890,7 +907,16 @@ def show_port_info(self, port_id: int) -> TestPmdPort:
if output.startswith("Invalid port"):
raise InteractiveCommandExecutionError("invalid port given")
- return TestPmdPort.parse(output)
+ port = TestPmdPort.parse(output)
+ self._update_port(port)
+ return port
+
+ def _update_port(self, port: TestPmdPort) -> None:
+ if self._ports:
+ self._ports = [
+ existing_port if port.id != existing_port.id else port
+ for existing_port in self._ports
+ ]
def show_port_stats_all(self) -> list[TestPmdPortStats]:
"""Returns the statistics of all the ports.
--
2.34.1
