[PATCH 0/2] dts: add packet capture test suite

[PATCH 0/2] dts: add packet capture test suite

[Thomas Wilks]

Hi,

Sending this new test suite that tests the packet capture framework.

Best regards,
Thomas

Depends-on: series-34865 ("dts: shell improvements")

Thomas Wilks (2):
  dts: add packet capture test suite
  dts: import lldp package in scapy

 .../dts/tests.TestSuite_packet_capture.rst    |   8 +
 .../testbed_model/traffic_generator/scapy.py  |   1 +
 dts/tests/TestSuite_packet_capture.py         | 357 ++++++++++++++++++
 3 files changed, 366 insertions(+)
 create mode 100644 doc/api/dts/tests.TestSuite_packet_capture.rst
 create mode 100644 dts/tests/TestSuite_packet_capture.py

-- 
2.43.0

[PATCH 1/2] dts: add packet capture test suite

[Thomas Wilks]

Add a test suite that tests the packet capture framework
through the use of dpdk-pdump.

Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
Reviewed-by: Luca Vizzarro <luca.vizzarro@arm.com>
---
 .../dts/tests.TestSuite_packet_capture.rst    |   8 +
 dts/tests/TestSuite_packet_capture.py         | 358 ++++++++++++++++++
 2 files changed, 366 insertions(+)
 create mode 100644 doc/api/dts/tests.TestSuite_packet_capture.rst
 create mode 100644 dts/tests/TestSuite_packet_capture.py

diff --git a/doc/api/dts/tests.TestSuite_packet_capture.rst b/doc/api/dts/tests.TestSuite_packet_capture.rst
new file mode 100644
index 0000000000..3d760d3ae4
--- /dev/null
+++ b/doc/api/dts/tests.TestSuite_packet_capture.rst
@@ -0,0 +1,8 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+
+packet_capture Test Suite
+=========================
+
+.. automodule:: tests.TestSuite_packet_capture
+   :members:
+   :show-inheritance:
diff --git a/dts/tests/TestSuite_packet_capture.py b/dts/tests/TestSuite_packet_capture.py
new file mode 100644
index 0000000000..bdc3d008a3
--- /dev/null
+++ b/dts/tests/TestSuite_packet_capture.py
@@ -0,0 +1,358 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 Arm Limited
+
+"""Packet capture TestSuite.
+
+Tests pdump by sending packets, changing the arguments of pdump
+and verifying the received packets.
+"""
+
+from dataclasses import asdict, dataclass
+from pathlib import Path, PurePath
+from random import randint
+from typing import Union
+
+from scapy.contrib.lldp import (
+    LLDPDUChassisID,
+    LLDPDUEndOfLLDPDU,
+    LLDPDUPortID,
+    LLDPDUSystemCapabilities,
+    LLDPDUSystemDescription,
+    LLDPDUSystemName,
+    LLDPDUTimeToLive,
+)
+from scapy.layers.inet import IP, TCP, UDP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Dot1Q, Ether
+from scapy.layers.sctp import SCTP
+from scapy.packet import Packet, Raw
+from scapy.utils import rdpcap
+from typing_extensions import Literal
+
+from framework.params.eal import EalParams
+from framework.params.testpmd import PortTopology
+from framework.remote_session.dpdk_app import BlockingDPDKApp
+from framework.remote_session.testpmd_shell import TestPmdShell
+from framework.settings import SETTINGS
+from framework.test_suite import TestSuite, func_test
+from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
+    PacketFilteringConfig,
+)
+
+
+@dataclass(order=True, kw_only=True)
+class PdumpParameters:
+    """Parameters for --pdump.
+
+    Attributes:
+        device_id: The PCIE address of the device to capture packets.
+        port: The port id of the device to capture packets.
+        queue: The ID of the RX and TX queue.
+        tx_dev: The path or iface that the TX packets captured on TX will be outputted at.
+        rx_dev: The path or iface that the RX packets captured on RX will be outputted at.
+        ring_size: The size of the ring.
+        mbuf_size: The size of the mbuf data.
+        total_num_mbufs: Total number of the mbufs in the memory pool.
+    """
+
+    device_id: str | None = None
+    port: int | None = None
+    queue: Union[int, Literal["*"]] = "*"
+    tx_dev: PurePath | str | None = None
+    rx_dev: PurePath | str | None = None
+    ring_size: int | None = None
+    mbuf_size: int | None = None
+    total_num_mbufs: int | None = None
+
+    def __str__(self) -> str:
+        """Parameters as string for --pdump.
+
+        Returns:
+            String of the pdump commands
+        """
+
+        def pair_to_str(field, value):
+            if field != "device_id":
+                field = field.replace("_", "-")
+            return f"{field}={value}"
+
+        arg = ",".join(
+            [
+                pair_to_str(field, value)
+                for field, value in asdict(self).items()
+                if value is not None
+            ]
+        )
+
+        return arg
+
+
+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.
+        pdump_params: The list of pdump parameters.
+    """
+
+    packets: list[Packet]
+    rx_pcap_path: PurePath
+    tx_pcap_path: PurePath
+    pdump_params: list[PdumpParameters]
+
+    def set_up_suite(self) -> None:
+        """Test suite setup.
+
+        Prepare the packets, file paths and queue range to be used in the test suite.
+        """
+        self.packets = [
+            Ether() / IP() / Raw(b"\0" * 60),
+            Ether() / IP() / TCP() / Raw(b"\0" * 60),
+            Ether() / IP() / UDP() / Raw(b"\0" * 60),
+            Ether() / IP() / SCTP() / Raw(b"\0" * 40),
+            Ether() / IPv6() / TCP() / Raw(b"\0" * 60),
+            Ether() / IPv6() / UDP() / Raw(b"\0" * 60),
+            Ether() / IP() / IPv6() / SCTP() / Raw(b"\0" * 40),
+            Ether() / Dot1Q() / IP() / UDP() / Raw(b"\0" * 40),
+            Ether(dst="FF:FF:FF:FF:FF:FF", type=0x88F7) / Raw(b"\0" * 60),
+            Ether(type=0x88CC)
+            / LLDPDUChassisID(subtype=4, id=self.topology.tg_port_egress.mac_address)
+            / LLDPDUPortID(subtype=5, id="Test Id")
+            / LLDPDUTimeToLive(ttl=180)
+            / LLDPDUSystemName(system_name="DTS Test sys")
+            / LLDPDUSystemDescription(description="DTS Test Packet")
+            / LLDPDUSystemCapabilities()
+            / LLDPDUEndOfLLDPDU(),
+        ]
+        self.tx_pcap_path = self._ctx.sut_node.tmp_dir.joinpath("tx.pcap")
+        self.rx_pcap_path = self._ctx.sut_node.tmp_dir.joinpath("rx.pcap")
+
+    def set_up_test_case(self) -> None:
+        """Set up test case.
+
+        Prepare default pdump parameters.
+        """
+        self.pdump_params = [
+            PdumpParameters(
+                device_id=self.topology.sut_port_ingress.pci,
+                rx_dev=self.rx_pcap_path,
+            ),
+            PdumpParameters(
+                device_id=self.topology.sut_port_egress.pci,
+                tx_dev=self.tx_pcap_path,
+            ),
+        ]
+
+    def _load_pcap_packets(self, remote_pcap_path: PurePath) -> list[Packet]:
+        local_pcap_path = Path(SETTINGS.output_dir).joinpath(remote_pcap_path.name)
+        self._ctx.sut_node.main_session.copy_from(remote_pcap_path, local_pcap_path)
+        return list(rdpcap(str(local_pcap_path)))
+
+    def _verify_packets(self, received_packets: list[Packet]) -> None:
+        expected_packets = self.get_expected_packets(self.packets, sent_from_tg=True)
+        rx_pcap_packets = self._load_pcap_packets(self.rx_pcap_path)
+        self.verify(
+            self.match_all_packets(expected_packets, rx_pcap_packets, verify=False),
+            "RX packet from pdump wasn't the same as the packet from Scapy.",
+        )
+
+        tx_pcap_packets = self._load_pcap_packets(self.tx_pcap_path)
+        self.verify(
+            self.match_all_packets(tx_pcap_packets, received_packets, verify=False),
+            "TX packet from pdump wasn't the same as the expected packet.",
+        )
+
+    def _start_pdump(
+        self,
+        pdump_params: list[PdumpParameters],
+        eal_params: EalParams | None = None,
+    ) -> BlockingDPDKApp:
+        if eal_params is None:
+            eal_params = EalParams()
+
+        eal_params.append_str(" ".join([f'--pdump "{param}"' for param in pdump_params]))
+
+        pdump = BlockingDPDKApp(PurePath("app/dpdk-pdump"), app_params=eal_params)
+        pdump.wait_until_ready(
+            "queue 65535" if pdump_params[0].queue == "*" else f"queue {pdump_params[0].queue}"
+        )
+        return pdump
+
+    def _start_pdump_and_verify_packets(self) -> None:
+        pdump = self._start_pdump(self.pdump_params)
+        received_packets = self.send_packets_and_capture(
+            self.packets, PacketFilteringConfig(no_lldp=False)
+        )
+        pdump.close()
+        self._verify_packets(received_packets)
+
+    @func_test
+    def test_pdump_device_id(self) -> None:
+        """Test pdump device id.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump with the default values.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+            self._start_pdump_and_verify_packets()
+
+    @func_test
+    def test_pdump_port(self) -> None:
+        """Test pdump port.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump specifying port numbers instead of device ids.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+
+            self.pdump_params[0].port = 0
+            self.pdump_params[0].device_id = None
+            self.pdump_params[1].port = 1
+            self.pdump_params[1].device_id = None
+
+            self._start_pdump_and_verify_packets()
+
+    @func_test
+    def test_pdump_queue(self) -> None:
+        """Test pdump queue.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump specifying queue 0 instead of the default queue value.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+
+            self.pdump_params[0].queue = 0
+            self.pdump_params[1].queue = 0
+
+            self._start_pdump_and_verify_packets()
+
+    @func_test
+    def test_pdump_dev_iface(self) -> None:
+        """Test pdump dev iface.
+
+        Dump RX packets to the link that scapy is capturing packets on.
+
+        Steps:
+            Switch the SUT egress port driver to the kernel driver and bring up the device.
+            Start up TestPMD Shell with loop port topology.
+            Boot up Pdump with the SUT port egress logical name instead of the pcap file.
+            Send packet and capture.
+
+        Verify:
+            Verify the expected packets are the same as the packets received by Scapy.
+        """
+        try:
+            self._ctx.dpdk.bind_ports_to_driver(self.topology.sut_ports[1:], for_dpdk=False)
+            self._ctx.sut_node.main_session.bring_up_link(self.topology.sut_ports[1:])
+
+            with TestPmdShell(
+                allowed_ports=self.topology.sut_ports[:1], port_topology=PortTopology.loop
+            ) as testpmd:
+                testpmd.start()
+
+                pdump = self._start_pdump(
+                    eal_params=EalParams(allowed_ports=self.topology.sut_ports[:1]),
+                    pdump_params=[
+                        PdumpParameters(
+                            device_id=self.topology.sut_port_ingress.pci,
+                            rx_dev=self.topology.sut_port_egress.logical_name,
+                        ),
+                    ],
+                )
+                received_packets = self.send_packets_and_capture(
+                    self.packets, PacketFilteringConfig(no_lldp=False)
+                )
+                pdump.close()
+
+                self.match_all_packets(
+                    self.get_expected_packets(self.packets, sent_from_tg=True), received_packets
+                )
+        finally:
+            self._ctx.dpdk.bind_ports_to_driver(self.topology.sut_ports[1:], for_dpdk=True)
+
+    @func_test
+    def test_pdump_ring_size(self) -> None:
+        """Test pdump mbuf size.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump with a custom ring size of 1024.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+
+            self.pdump_params[0].ring_size = 1024
+            self.pdump_params[1].ring_size = 1024
+
+            self._start_pdump_and_verify_packets()
+
+    @func_test
+    def test_pdump_mbuf_size(self) -> None:
+        """Test pdump total num mbufs.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump with a custom mbuf size of 2048.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+
+            self.pdump_params[0].mbuf_size = 2048
+            self.pdump_params[1].mbuf_size = 2048
+
+            self._start_pdump_and_verify_packets()
+
+    @func_test
+    def test_pdump_total_num_mbufs(self) -> None:
+        """Test pdump total num mbufs.
+
+        Steps:
+            Start up TestPMD Shell.
+            Boot up Pdump with a custom total num mbufs value chosen at random from 1025 to 65535.
+            Send packets.
+
+        Verify:
+            Verify the expected packets are the same as the RX packets.
+            Verify the TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+
+            total_num_mbufs = randint(1025, 65535)
+            self.pdump_params[0].total_num_mbufs = total_num_mbufs
+            self.pdump_params[1].total_num_mbufs = total_num_mbufs
+
+            self._start_pdump_and_verify_packets()
-- 
2.43.0

Re: [PATCH 1/2] dts: add packet capture test suite

[Dean Marx]

On Mon, Mar 31, 2025 at 11:58 AM Thomas Wilks <thomas.wilks@arm.com> wrote:
>
> Add a test suite that tests the packet capture framework
> through the use of dpdk-pdump.
>
> Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
> Reviewed-by: Luca Vizzarro <luca.vizzarro@arm.com>
> ---
>  .../dts/tests.TestSuite_packet_capture.rst    |   8 +
>  dts/tests/TestSuite_packet_capture.py         | 358 ++++++++++++++++++
>  2 files changed, 366 insertions(+)
>  create mode 100644 doc/api/dts/tests.TestSuite_packet_capture.rst
>  create mode 100644 dts/tests/TestSuite_packet_capture.py

<snip>

> +    device_id: str | None = None
> +    port: int | None = None
> +    queue: Union[int, Literal["*"]] = "*"
> +    tx_dev: PurePath | str | None = None
> +    rx_dev: PurePath | str | None = None
> +    ring_size: int | None = None
> +    mbuf_size: int | None = None
> +    total_num_mbufs: int | None = None

One small thing I noticed here is that you're using Union for the
queue and pipes for the rest of the declarations. queue: int |
Literal["*"] = "*" would be the same thing functionally, and it would
save you an import.

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

Re: [PATCH 1/2] dts: add packet capture test suite

[Stephen Hemminger]

On Mon, 31 Mar 2025 16:57:59 +0100
Thomas Wilks <thomas.wilks@arm.com> wrote:

> +    def _start_pdump(
> +        self,
> +        pdump_params: list[PdumpParameters],
> +        eal_params: EalParams | None = None,
> +    ) -> BlockingDPDKApp:
> +        if eal_params is None:
> +            eal_params = EalParams()
> +
> +        eal_params.append_str(" ".join([f'--pdump "{param}"' for param in pdump_params]))
> +
> +        pdump = BlockingDPDKApp(PurePath("app/dpdk-pdump"), app_params=eal_params)

Please test dpdk-dumpcap not pdump. The old pdump is legacy, has too complex an options,
can't work in many cases and is a general mess. Dpdk-dumpcap was built as a replacement
and has way more features and will be supported.

I intend to deprecate pdump and remove it in 25.11.

[PATCH 2/2] dts: import lldp package in scapy

[Thomas Wilks]

Add import for lldp scapy package to enable lldp packet
creation and handling.

Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
Reviewed-by: Luca Vizzarro <luca.vizzarro@arm.com>
---
 dts/framework/testbed_model/traffic_generator/scapy.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
index 78a6ded74c..c7e8fc42e9 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -95,6 +95,7 @@ def setup(self, ports: Iterable[Port]):
         self._tg_node.main_session.bring_up_link(ports)
         self._shell.start_application()
         self._shell.send_command("from scapy.all import *")
+        self._shell.send_command("from scapy.contrib.lldp import *")
 
     def close(self):
         """Close traffic generator."""
-- 
2.43.0

Re: [PATCH 2/2] dts: import lldp package in scapy

[Dean Marx]

On Mon, Mar 31, 2025 at 11:58 AM Thomas Wilks <thomas.wilks@arm.com> wrote:
>
> Add import for lldp scapy package to enable lldp packet
> creation and handling.
>
> Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
> Reviewed-by: Luca Vizzarro <luca.vizzarro@arm.com>

Reviewed-by: Dean Marx <dmarx@iol.unh.edu>

[PATCH v2 0/3] dts: add packet capture test suite

[Luca Vizzarro]

Hi there,

sending in a v2 for the packet capture test suite.

v2:
- added a new command to change file permissions on remote
- changed test suite to use dpdk-dumpcap instead of dpdk-pdump

Best regards,
Luca

Luca Vizzarro (1):
  dts: add facility to change file permissions

Thomas Wilks (2):
  dts: import lldp package in scapy
  dts: add packet capture test suite

 .../dts/tests.TestSuite_packet_capture.rst    |   8 +
 dts/framework/testbed_model/os_session.py     |  34 +++
 dts/framework/testbed_model/posix_session.py  |  11 +-
 .../testbed_model/traffic_generator/scapy.py  |   1 +
 dts/tests/TestSuite_packet_capture.py         | 238 ++++++++++++++++++
 5 files changed, 291 insertions(+), 1 deletion(-)
 create mode 100644 doc/api/dts/tests.TestSuite_packet_capture.rst
 create mode 100644 dts/tests/TestSuite_packet_capture.py

-- 
2.43.0

[PATCH v2 1/3] dts: import lldp package in scapy

[Luca Vizzarro]

From: Thomas Wilks <thomas.wilks@arm.com>

Add import for lldp scapy package to enable lldp packet
creation and handling.

Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepanek@arm.com>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/testbed_model/traffic_generator/scapy.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py b/dts/framework/testbed_model/traffic_generator/scapy.py
index aed0b76108..76f8496bf8 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -328,6 +328,7 @@ def setup(self, ports: Iterable[Port], rx_port: Port):
         self._shell = PythonShell(self._tg_node, "scapy", privileged=True)
         self._shell.start_application()
         self._shell.send_command("from scapy.all import *")
+        self._shell.send_command("from scapy.contrib.lldp import *")
 
     def close(self):
         """Overrides :meth:`.traffic_generator.TrafficGenerator.close`.
-- 
2.43.0

[PATCH v2 2/3] dts: add facility to change file permissions

[Luca Vizzarro]

Add an abstraction to handle file permissions for the remote.

Signed-off-by: Luca Vizzarro <luca.vizzarro@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepanek@arm.com>
---
 dts/framework/testbed_model/os_session.py    | 34 ++++++++++++++++++++
 dts/framework/testbed_model/posix_session.py | 11 ++++++-
 2 files changed, 44 insertions(+), 1 deletion(-)

diff --git a/dts/framework/testbed_model/os_session.py b/dts/framework/testbed_model/os_session.py
index 354c607357..869d575f0e 100644
--- a/dts/framework/testbed_model/os_session.py
+++ b/dts/framework/testbed_model/os_session.py
@@ -26,6 +26,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import Iterable
 from dataclasses import dataclass
+from enum import Flag, auto
 from pathlib import Path, PurePath, PurePosixPath
 
 from framework.config.node import NodeConfiguration
@@ -44,6 +45,33 @@
 from .port import Port
 
 
+class FilePermissions(Flag):
+    """The permissions for a file and/or directory."""
+
+    #:
+    OTHERS_EXECUTE = auto()
+    #:
+    OTHERS_WRITE = auto()
+    #:
+    OTHERS_READ = auto()
+    #:
+    GROUP_EXECUTE = auto()
+    #:
+    GROUP_WRITE = auto()
+    #:
+    GROUP_READ = auto()
+    #:
+    OWNER_EXECUTE = auto()
+    #:
+    OWNER_WRITE = auto()
+    #:
+    OWNER_READ = auto()
+
+    def to_octal(self) -> str:
+        """Convert this flag to an octal representation."""
+        return format(self.value, "03o")
+
+
 @dataclass(slots=True, frozen=True)
 class OSSessionInfo:
     """Supplemental OS session information.
@@ -313,6 +341,12 @@ def copy_dir_to(
                 These patterns are used with `fnmatch.fnmatch` to filter out files.
         """
 
+    @abstractmethod
+    def change_permissions(
+        self, remote_path: PurePath, permissions: FilePermissions, recursive: bool = False
+    ) -> None:
+        """Change the permissions of the given path."""
+
     @abstractmethod
     def remove_remote_file(self, remote_file_path: str | PurePath, force: bool = True) -> None:
         """Remove remote file, by default remove forcefully.
diff --git a/dts/framework/testbed_model/posix_session.py b/dts/framework/testbed_model/posix_session.py
index 2d2701e1cf..fcd2faad6b 100644
--- a/dts/framework/testbed_model/posix_session.py
+++ b/dts/framework/testbed_model/posix_session.py
@@ -26,7 +26,7 @@
 )
 
 from .cpu import Architecture
-from .os_session import OSSession, OSSessionInfo
+from .os_session import FilePermissions, OSSession, OSSessionInfo
 
 
 class PosixSession(OSSession):
@@ -146,6 +146,15 @@ def copy_dir_to(
         self.extract_remote_tarball(remote_tar_path)
         self.remove_remote_file(remote_tar_path)
 
+    def change_permissions(
+        self, remote_path: PurePath, permissions: FilePermissions, recursive: bool = False
+    ) -> None:
+        """Overrides :meth:`~.os_session.OSSession.change_permissions`."""
+        self.send_command(
+            f"chmod {'-R ' if recursive else ''}{permissions.to_octal()} {remote_path}",
+            privileged=True,
+        )
+
     def remove_remote_file(self, remote_file_path: str | PurePath, force: bool = True) -> None:
         """Overrides :meth:`~.os_session.OSSession.remove_remote_dir`."""
         opts = PosixSession.combine_short_options(f=force)
-- 
2.43.0

[PATCH v2 3/3] dts: add packet capture test suite

[Luca Vizzarro]

From: Thomas Wilks <thomas.wilks@arm.com>

Add a test suite that tests the packet capture framework
through the use of dpdk-dumpcap.

Signed-off-by: Thomas Wilks <thomas.wilks@arm.com>
Signed-off-by: Luca Vizzarro <luca.vizzarro@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepanek@arm.com>
---
 .../dts/tests.TestSuite_packet_capture.rst    |   8 +
 dts/tests/TestSuite_packet_capture.py         | 238 ++++++++++++++++++
 2 files changed, 246 insertions(+)
 create mode 100644 doc/api/dts/tests.TestSuite_packet_capture.rst
 create mode 100644 dts/tests/TestSuite_packet_capture.py

diff --git a/doc/api/dts/tests.TestSuite_packet_capture.rst b/doc/api/dts/tests.TestSuite_packet_capture.rst
new file mode 100644
index 0000000000..3d760d3ae4
--- /dev/null
+++ b/doc/api/dts/tests.TestSuite_packet_capture.rst
@@ -0,0 +1,8 @@
+.. SPDX-License-Identifier: BSD-3-Clause
+
+packet_capture Test Suite
+=========================
+
+.. automodule:: tests.TestSuite_packet_capture
+   :members:
+   :show-inheritance:
diff --git a/dts/tests/TestSuite_packet_capture.py b/dts/tests/TestSuite_packet_capture.py
new file mode 100644
index 0000000000..67a3049b78
--- /dev/null
+++ b/dts/tests/TestSuite_packet_capture.py
@@ -0,0 +1,238 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2025 Arm Limited
+
+"""Packet capture test suite.
+
+Test the DPDK packet capturing framework through the combined use of testpmd and dumpcap.
+"""
+
+from dataclasses import dataclass, field
+from pathlib import Path, PurePath
+
+from scapy.contrib.lldp import (
+    LLDPDUChassisID,
+    LLDPDUEndOfLLDPDU,
+    LLDPDUPortID,
+    LLDPDUSystemCapabilities,
+    LLDPDUSystemDescription,
+    LLDPDUSystemName,
+    LLDPDUTimeToLive,
+)
+from scapy.layers.inet import IP, TCP, UDP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Dot1Q, Ether
+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.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
+from framework.testbed_model.capability import requires
+from framework.testbed_model.cpu import LogicalCoreList
+from framework.testbed_model.os_session import FilePermissions
+from framework.testbed_model.topology import TopologyType
+from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
+    PacketFilteringConfig,
+)
+
+
+@dataclass(kw_only=True)
+class DumpcapParams(Params):
+    """Parameters for the dpdk-dumpcap app.
+
+    Attributes:
+        lcore_list: The list of logical cores to use.
+        file_prefix: The DPDK file prefix that the primary DPDK application is using.
+        interface: The PCI address of the interface to capture.
+        output_pcap_path: The path to the pcapng file to dump captured packets into.
+        packet_filter: A packet filter in libpcap filter syntax.
+    """
+
+    lcore_list: LogicalCoreList | None = field(default=None, metadata=Params.long("lcore"))
+    file_prefix: str | None = None
+    interface: str = field(metadata=Params.short("i"))
+    output_pcap_path: PurePath = field(metadata=Params.convert_value(str) | Params.short("w"))
+    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[Packet]
+    rx_pcap_path: PurePath
+    tx_pcap_path: PurePath
+
+    def set_up_suite(self) -> None:
+        """Test suite setup.
+
+        Prepare the packets, file paths and queue range to be used in the test suite.
+        """
+        self.packets = [
+            Ether() / IP() / Raw(b"\0" * 60),
+            Ether() / IP() / TCP() / Raw(b"\0" * 60),
+            Ether() / IP() / UDP() / Raw(b"\0" * 60),
+            Ether() / IP() / SCTP() / Raw(b"\0" * 40),
+            Ether() / IPv6() / TCP() / Raw(b"\0" * 60),
+            Ether() / IPv6() / UDP() / Raw(b"\0" * 60),
+            Ether() / IP() / IPv6() / SCTP() / Raw(b"\0" * 40),
+            Ether() / Dot1Q() / IP() / UDP() / Raw(b"\0" * 40),
+            Ether(dst="FF:FF:FF:FF:FF:FF", type=0x88F7) / Raw(b"\0" * 60),
+            Ether(type=0x88CC)
+            / LLDPDUChassisID(subtype=4, id=self.topology.tg_port_egress.mac_address)
+            / LLDPDUPortID(subtype=5, id="Test Id")
+            / LLDPDUTimeToLive(ttl=180)
+            / LLDPDUSystemName(system_name="DTS Test sys")
+            / LLDPDUSystemDescription(description="DTS Test Packet")
+            / LLDPDUSystemCapabilities()
+            / LLDPDUEndOfLLDPDU(),
+        ]
+        self.tx_pcap_path = self._ctx.sut_node.tmp_dir.joinpath("tx.pcapng")
+        self.rx_pcap_path = self._ctx.sut_node.tmp_dir.joinpath("rx.pcapng")
+
+    def _load_pcap_packets(self, remote_pcap_path: PurePath) -> list[Packet]:
+        local_pcap_path = Path(SETTINGS.output_dir).joinpath(remote_pcap_path.name)
+        self._ctx.sut_node.main_session.copy_from(remote_pcap_path, local_pcap_path)
+        return list(rdpcap(str(local_pcap_path)))
+
+    def _send_and_dump(
+        self, packet_filter: str | None = None, rx_only: bool = False
+    ) -> list[Packet]:
+        dumpcap_rx = 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(
+                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)
+        )
+
+        dumpcap_rx.close()
+        self._ctx.sut_node.main_session.change_permissions(
+            self.rx_pcap_path, FilePermissions(0o644)
+        )
+        if not rx_only:
+            dumpcap_tx.close()
+            self._ctx.sut_node.main_session.change_permissions(
+                self.tx_pcap_path, FilePermissions(0o644)
+            )
+
+        return received_packets
+
+    @func_test
+    def test_dumpcap(self) -> None:
+        """Test dumpcap on RX and TX interfaces.
+
+        Steps:
+            * Start up testpmd shell.
+            * Start up dpdk-dumpcap with the default values.
+            * Send packets.
+
+        Verify:
+            * The expected packets are the same as the RX packets.
+            * The TX packets are the same as the packets received from Scapy.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+            received_packets = self._send_and_dump()
+
+            expected_packets = self.get_expected_packets(self.packets, sent_from_tg=True)
+            rx_pcap_packets = self._load_pcap_packets(self.rx_pcap_path)
+            self.verify(
+                self.match_all_packets(expected_packets, rx_pcap_packets, verify=False),
+                "RX packets from dumpcap weren't the same as the expected packets.",
+            )
+
+            tx_pcap_packets = self._load_pcap_packets(self.tx_pcap_path)
+            self.verify(
+                self.match_all_packets(tx_pcap_packets, received_packets, verify=False),
+                "TX packets from dumpcap weren't the same as the packets received by Scapy.",
+            )
+
+    @func_test
+    def test_dumpcap_filter(self) -> None:
+        """Test the dumpcap filtering feature.
+
+        Steps:
+            * Start up testpmd shell.
+            * Start up dpdk-dumpcap listening for TCP packets on the RX interface.
+            * Send packets.
+
+        Verify:
+            * The dumped packets did not contain any of the packets meant for filtering.
+        """
+        with TestPmdShell() as testpmd:
+            testpmd.start()
+            self._send_and_dump("tcp", rx_only=True)
+            filtered_packets = [
+                raw(p)
+                for p in self.get_expected_packets(self.packets, sent_from_tg=True)
+                if not p.haslayer(TCP)
+            ]
+
+            rx_pcap_packets = [raw(p) for p in self._load_pcap_packets(self.rx_pcap_path)]
+            for filtered_packet in filtered_packets:
+                self.verify(
+                    filtered_packet not in rx_pcap_packets,
+                    "Found a packet in the pcap that was meant to be filtered out.",
+                )
-- 
2.43.0