Add a generic blocking app class to allow a test writer to run any app
in the background. Make the original BlockingDPDKApp class inherit from
this one. Implement the new BlockingApp class in the packet_capture test
suite.
Signed-off-by: Luca Vizzarro <luca.vizza...@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepa...@arm.com>
---
dts/framework/remote_session/blocking_app.py | 135 +++++++++++++++++++
dts/framework/remote_session/dpdk_app.py | 80 -----------
dts/tests/TestSuite_packet_capture.py | 66 +++------
3 files changed, 154 insertions(+), 127 deletions(-)
create mode 100644 dts/framework/remote_session/blocking_app.py
delete mode 100644 dts/framework/remote_session/dpdk_app.py
diff --git a/dts/framework/remote_session/blocking_app.py
b/dts/framework/remote_session/blocking_app.py
new file mode 100644
index 0000000000..8de536c259
--- /dev/null
+++ b/dts/framework/remote_session/blocking_app.py
@@ -0,0 +1,135 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 Arm Limited
+
+"""Class to run blocking apps in the background.
+
+The class won't automatically start the app. The start-up is done as part of
the
+:meth:`BlockingApp.wait_until_ready` method, which will return execution to
the caller only
+when the desired stdout has been returned by the app. Usually this is used to
detect when the app
+has been loaded and ready to be used.
+
+This module also provides the class :class:`BlockingDPDKApp` useful to run any
DPDK app from the
+DPDK build dir.
+
+Example:
+ ..code:: python
+
+ pdump = BlockingDPDKApp(
+ PurePath("app/dpdk-pdump"),
+ app_params="--pdump 'port=0,queue=*,rx-dev=/tmp/rx-dev.pcap'"
+ )
+ pdump.wait_until_ready("65535") # start app
+
+ # pdump is now ready to capture
+
+ pdump.close() # stop/close app
+"""
+
+from pathlib import PurePath
+from typing import Generic, TypeVar, cast
+
+from typing_extensions import Self
+
+from framework.context import get_ctx
+from framework.params import Params
+from framework.params.eal import EalParams
+from framework.remote_session.dpdk_shell import compute_eal_params
+from framework.remote_session.interactive_shell import InteractiveShell
+from framework.testbed_model.node import Node
+
+P = TypeVar("P", bound=Params)
+
+
+class BlockingApp(InteractiveShell, Generic[P]):
+ """Class to manage generic blocking apps."""
+
+ _app_params: P
+
+ def __init__(
+ self,
+ node: Node,
+ path: PurePath,
+ name: str | None = None,
+ privileged: bool = False,
+ app_params: P | str = "",
+ ) -> None:
+ """Constructor.
+
+ Args:
+ node: The node to run the app on.
+ path: Path to the application on the node.
+ name: Name to identify this application.
+ privileged: Run as privileged user.
+ app_params: The application parameters. Can be of any type
inheriting :class:`Params` or
+ a plain string.
+ """
+ if isinstance(app_params, str):
+ params = Params()
+ params.append_str(app_params)
+ app_params = cast(P, params)
+
+ self._path = path
+
+ super().__init__(node, name, privileged, app_params)
+
+ @property
+ def path(self) -> PurePath:
+ """The path of the DPDK app relative to the DPDK build folder."""
+ return self._path
+
+ def wait_until_ready(self, end_token: str) -> Self:
+ """Start app and wait until ready.
+
+ Args:
+ end_token: The string at the end of a line that indicates the app
is ready.
+
+ Returns:
+ Itself.
+ """
+ self.start_application(end_token)
+ return self
+
+ def close(self) -> None:
+ """Close the application.
+
+ Sends a SIGINT to close the application.
+ """
+ self.send_command("\x03")
+ super().close()
+
+
+PE = TypeVar("PE", bound=EalParams)
+
+
+class BlockingDPDKApp(BlockingApp, Generic[PE]):
+ """Class to manage blocking DPDK apps on the SUT."""
+
+ _app_params: PE
+
+ def __init__(
+ self,
+ path: PurePath,
+ name: str | None = None,
+ privileged: bool = True,
+ app_params: PE | str = "",
+ ) -> None:
+ """Constructor.
+
+ Args:
+ path: Path relative to the DPDK build to the executable.
+ name: Name to identify this application.
+ privileged: Run as privileged user.
+ app_params: The application parameters. If a string or an
incomplete :class:`EalParams`
+ object are passed, the EAL params are computed based on the
current context.
+ """
+ if isinstance(app_params, str):
+ eal_params = compute_eal_params()
+ eal_params.append_str(app_params)
+ app_params = cast(PE, eal_params)
+ else:
+ app_params = cast(PE, compute_eal_params(app_params))
+
+ node = get_ctx().sut_node
+ path =
PurePath(get_ctx().dpdk_build.remote_dpdk_build_dir).joinpath(self.path)
+
+ super().__init__(node, path, name, privileged, app_params)
diff --git a/dts/framework/remote_session/dpdk_app.py
b/dts/framework/remote_session/dpdk_app.py
deleted file mode 100644
index dc4b817bdd..0000000000
--- a/dts/framework/remote_session/dpdk_app.py
+++ /dev/null
@@ -1,80 +0,0 @@
-# SPDX-License-Identifier: BSD-3-Clause
-# Copyright(c) 2025 Arm Limited
-
-"""Class to run blocking DPDK apps in the background.
-
-The class won't automatically start the app. The start-up is done as part of
the
-:meth:`BlockingDPDKApp.wait_until_ready` method, which will return execution
to the caller only
-when the desired stdout has been returned by the app. Usually this is used to
detect when the app
-has been loaded and ready to be used.
-
-Example:
- ..code:: python
-
- pdump = BlockingDPDKApp(
- PurePath("app/dpdk-pdump"),
- app_params="--pdump 'port=0,queue=*,rx-dev=/tmp/rx-dev.pcap'"
- )
- pdump.wait_until_ready("65535") # start app
-
- # pdump is now ready to capture
-
- pdump.close() # stop/close app
-"""
-
-from pathlib import PurePath
-
-from framework.params.eal import EalParams
-from framework.remote_session.dpdk_shell import DPDKShell
-
-
-class BlockingDPDKApp(DPDKShell):
- """Class to manage blocking DPDK apps."""
-
- def __init__(
- self,
- path: PurePath,
- name: str | None = None,
- privileged: bool = True,
- app_params: EalParams | str = "",
- ) -> None:
- """Constructor.
-
- Overrides :meth:`~.dpdk_shell.DPDKShell.__init__`.
-
- Args:
- path: Path relative to the DPDK build to the executable.
- name: Name to identify this application.
- privileged: Run as privileged user.
- app_params: The application parameters. If a string or an
incomplete :class:`EalParams`
- object are passed, the EAL params are computed based on the
current context.
- """
- if isinstance(app_params, str):
- eal_params = EalParams()
- eal_params.append_str(app_params)
- app_params = eal_params
-
- self._path = path
-
- super().__init__(name, privileged, app_params)
-
- @property
- def path(self) -> PurePath:
- """The path of the DPDK app relative to the DPDK build folder."""
- return self._path
-
- def wait_until_ready(self, end_token: str) -> None:
- """Start app and wait until ready.
-
- Args:
- end_token: The string at the end of a line that indicates the app
is ready.
- """
- self.start_application(end_token)
-
- def close(self) -> None:
- """Close the application.
-
- Sends a SIGINT to close the application.
- """
- self.send_command("\x03")
- super().close()
diff --git a/dts/tests/TestSuite_packet_capture.py
b/dts/tests/TestSuite_packet_capture.py
index e162bded87..bad243a571 100644
--- a/dts/tests/TestSuite_packet_capture.py
+++ b/dts/tests/TestSuite_packet_capture.py
@@ -24,12 +24,10 @@
from scapy.layers.sctp import SCTP
from scapy.packet import Packet, Raw, raw
from scapy.utils import rdpcap
-from typing_extensions import Self
-from framework.context import get_ctx
from framework.params import Params
+from framework.remote_session.blocking_app import BlockingApp
from framework.remote_session.dpdk_shell import compute_eal_params
-from framework.remote_session.interactive_shell import InteractiveShell
from framework.remote_session.testpmd_shell import TestPmdShell
from framework.settings import SETTINGS
from framework.test_suite import TestSuite, func_test
@@ -61,57 +59,31 @@ class DumpcapParams(Params):
packet_filter: str | None = field(default=None, metadata=Params.short("f"))
-class Dumpcap(InteractiveShell):
- """Class to spawn and manage a dpdk-dumpcap process.
-
- The dpdk-dumpcap is a DPDK app but instead of providing a regular DPDK EAL
interface to the
- user, it replicates the Wireshark dumpcap app.
- """
-
- _app_params: DumpcapParams
-
- def __init__(self, params: DumpcapParams) -> None:
- """Extends :meth:`~.interactive_shell.InteractiveShell.__init__`."""
- self.ctx = get_ctx()
- eal_params = compute_eal_params()
- params.lcore_list = eal_params.lcore_list
- params.file_prefix = eal_params.prefix
-
- super().__init__(self.ctx.sut_node, name=None, privileged=True,
app_params=params)
-
- @property
- def path(self) -> PurePath:
- """Path to the shell executable."""
- return
PurePath(self.ctx.dpdk_build.remote_dpdk_build_dir).joinpath("app/dpdk-dumpcap")
-
- def wait_until_ready(self) -> Self:
- """Start app and wait until ready."""
- self.start_application(f"Capturing on '{self._app_params.interface}'")
- return self
-
- def close(self) -> None:
- """Close the application.
-
- Sends a SIGINT to close the application.
- """
- self.send_command("\x03")
- super().close()
-
-
@requires(topology_type=TopologyType.two_links)
class TestPacketCapture(TestSuite):
"""Packet Capture TestSuite.
Attributes:
- packets: List of packets to send for testing pdump.
- rx_pcap_path: The remote path where to create the Rx packets pcap with
pdump.
- tx_pcap_path: The remote path where to create the Tx packets pcap with
pdump.
+ packets: List of packets to send for testing dumpcap.
+ rx_pcap_path: The remote path where to create the Rx packets pcap with
dumpcap.
+ tx_pcap_path: The remote path where to create the Tx packets pcap with
dumpcap.
"""
packets: list[Packet]
rx_pcap_path: PurePath
tx_pcap_path: PurePath
+ def _run_dumpcap(self, params: DumpcapParams) -> BlockingApp:
+ eal_params = compute_eal_params()
+ params.lcore_list = eal_params.lcore_list
+ params.file_prefix = eal_params.prefix
+ return BlockingApp(
+ self._ctx.sut_node,
+ self._ctx.dpdk_build.get_app("dumpcap"),
+ app_params=params,
+ privileged=True,
+ ).wait_until_ready(f"Capturing on '{params.interface}'")
+
def set_up_suite(self) -> None:
"""Test suite setup.
@@ -147,21 +119,21 @@ def _load_pcap_packets(self, remote_pcap_path: PurePath)
-> list[Packet]:
def _send_and_dump(
self, packet_filter: str | None = None, rx_only: bool = False
) -> list[Packet]:
- dumpcap_rx = Dumpcap(
+ dumpcap_rx = self._run_dumpcap(
DumpcapParams(
interface=self.topology.sut_port_ingress.pci,
output_pcap_path=self.rx_pcap_path,
packet_filter=packet_filter,
)
- ).wait_until_ready()
+ )
if not rx_only:
- dumpcap_tx = Dumpcap(
+ dumpcap_tx = self._run_dumpcap(
DumpcapParams(
interface=self.topology.sut_port_egress.pci,
output_pcap_path=self.tx_pcap_path,
packet_filter=packet_filter,
)
- ).wait_until_ready()
+ )
received_packets = self.send_packets_and_capture(
self.packets, PacketFilteringConfig(no_lldp=False)
--
2.43.0
