On Fri, Oct 3, 2025 at 3:27 PM Dean Marx <dmarx@iol.unh.edu> wrote:

> Add test suite covering virtio-user and vhost
> server/client forwarding scenarios with
> testpmd packet validation.
>
> Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
> ---
>  doc/api/tests.TestSuite_virtio_fwd.rst |   8 ++
>  dts/tests/TestSuite_virtio_fwd.py      | 180 +++++++++++++++++++++++++
>  2 files changed, 188 insertions(+)
>  create mode 100644 doc/api/tests.TestSuite_virtio_fwd.rst
>  create mode 100644 dts/tests/TestSuite_virtio_fwd.py
>
> diff --git a/doc/api/tests.TestSuite_virtio_fwd.rst
> b/doc/api/tests.TestSuite_virtio_fwd.rst
> new file mode 100644
> index 0000000000..782eddad2d
> --- /dev/null
> +++ b/doc/api/tests.TestSuite_virtio_fwd.rst
> @@ -0,0 +1,8 @@
> +.. SPDX-License-Identifier: BSD-3-Clause
> +
> +virtio_fwd Test Suite
> +===========================
> +
> +.. automodule:: tests.TestSuite_virtio_fwd
> +   :members:
> +   :show-inheritance:
> \ No newline at end of file
> diff --git a/dts/tests/TestSuite_virtio_fwd.py
> b/dts/tests/TestSuite_virtio_fwd.py
> new file mode 100644
> index 0000000000..194bd24257
> --- /dev/null
> +++ b/dts/tests/TestSuite_virtio_fwd.py
> @@ -0,0 +1,180 @@
> +# SPDX-License-Identifier: BSD-3-Clause
> +# Copyright(c) 2025 University of New Hampshire
> +
> +"""Virtio forwarding test suite.
> +
> +Verify vhost/virtio pvp and loopback topology functionalities.
>

Loopback doesn't seem quite right. Maybe you can just say pvp and fully
virtual?


> +"""
> +
> +from scapy.layers.inet import IP
> +from scapy.layers.l2 import Ether
> +
> +from api.capabilities import LinkTopology
> +from api.testpmd import TestPmd
> +from api.testpmd.config import SimpleForwardingModes
> +from framework.parser import TextParser
> +from framework.test_suite import TestSuite, func_test
> +from framework.testbed_model.capability import requires
> +from framework.testbed_model.linux_session import LinuxSession
> +from framework.testbed_model.virtual_device import VirtualDevice
> +
> +
> +@requires(topology_type=LinkTopology.TWO_LINKS)
>

two links is true for the pvp test, but the other two dont require any
links, right?


> +class TestVirtioFwd(TestSuite):
> +    """Virtio forwarding test suite."""
> +
> +    class ForwardingParsers:
> +        """Class for gathering Rx/Tx packets from testpmd stats."""
> +
> +        rx_packets = TextParser.find_int(r"RX-packets:\s*(\d+)")
> +        tx_packets = TextParser.find_int(r"TX-packets:\s*(\d+)")
> +
> +    class vdevs:
> +        """Class containing virtio-user and vhost-user virtual devices."""
> +
> +        virtio_user = VirtualDevice(
> +
> "net_virtio_user0,mac=00:01:02:03:04:05,path=/tmp/vhost-net,server=1"
> +        )
> +        vhost_user =
> VirtualDevice("eth_vhost0,iface=/tmp/vhost-net,client=1")
>

+1 to Luca's comment about not needing these to be internal classes.


> +
> +    @func_test
> +    def virtio_server(self) -> None:
> +        """Test virtio server packet transmission.
> +
> +        Steps:
> +            * Launch a testpmd session with a vhost-user virtual device
> (client side).
> +            * Launch a testpmd session with a virtio-user virtual device
> (server side).
> +            * Set the forwarding mode to mac in both sessions.
> +            * Start packet forwarding on vhost session.
> +            * Send a burst of packets from the virtio session.
> +            * Stop packet forwarding on vhost session and collect Rx
> packet stats.
> +
> +        Verify:
> +            * Vhost session receives packets from virtio session.
> +        """
> +        with (
> +            TestPmd(
> +                prefix="vhost",
> +                no_pci=True,
> +                memory_channels=4,
> +                vdevs=[self.vdevs.vhost_user],
> +            ) as vhost,
> +            TestPmd(
> +                prefix="virtio",
> +                no_pci=True,
> +                memory_channels=4,
> +                vdevs=[self.vdevs.virtio_user],
> +            ) as virtio,
> +        ):
> +            vhost.set_forward_mode(SimpleForwardingModes.mac)
> +            virtio.set_forward_mode(SimpleForwardingModes.mac)
> +
> +            vhost.start()
> +            virtio.start_tx_first(burst_num=32)
> +
> +            forwarding_stats = vhost.stop()
> +
> +            rx_packets =
> self.ForwardingParsers.rx_packets["TextParser_fn"](forwarding_stats) or 0
> +            tx_packets =
> self.ForwardingParsers.tx_packets["TextParser_fn"](forwarding_stats) or 0
> +
> +            self.verify(
> +                rx_packets != 0 and tx_packets != 0,
> +                "Vhost session failed to receive packets from virtio
> session.",
> +            )
> +
> +    @func_test
> +    def virtio_server_reconnect(self) -> None:
> +        """Test virtio server reconnection.
> +
> +        Steps:
> +            * Launch a testpmd session with a vhost-user virtual device
> (client side).
> +            * Launch a testpmd session with a virtio-user virtual device
> (server side).
> +            * Close the virtio session and relaunch it.
> +            * Start packet forwarding on vhost session.
> +            * Send a burst of packets from the virtio session.
> +            * Stop packet forwarding on vhost session and collect Rx
> packet stats.
> +
> +        Verify:
> +            * Vhost session receives packets from relaunched virtio
> session.
> +        """
> +        with TestPmd(
> +            prefix="vhost",
> +            no_pci=True,
> +            memory_channels=4,
> +            vdevs=[self.vdevs.vhost_user],
> +        ) as vhost:
> +            with TestPmd(
> +                prefix="virtio",
> +                no_pci=True,
> +                memory_channels=4,
> +                vdevs=[self.vdevs.virtio_user],
> +            ) as virtio:
> +                pass
> +            # end session and reconnect
> +            with TestPmd(
> +                prefix="virtio",
> +                no_pci=True,
> +                memory_channels=4,
> +                vdevs=[self.vdevs.virtio_user],
> +            ) as virtio:
> +                virtio.set_forward_mode(SimpleForwardingModes.mac)
> +                vhost.set_forward_mode(SimpleForwardingModes.mac)
> +
> +                vhost.start()
> +                virtio.start_tx_first(burst_num=32)
> +
> +                forwarding_stats = vhost.stop()
> +
> +                rx_packets = (
> +
> self.ForwardingParsers.rx_packets["TextParser_fn"](forwarding_stats) or 0
> +                )
> +                tx_packets = (
> +
> self.ForwardingParsers.tx_packets["TextParser_fn"](forwarding_stats) or 0
> +                )
> +
> +                self.verify(
> +                    rx_packets != 0 and tx_packets != 0,
> +                    "Vhost session failed to receive packets from virtio
> session.",
> +                )
> +
> +    @func_test
> +    def pvp_loop(self) -> None:
> +        """Test vhost/virtio physical-virtual-physical loop topology.
> +
> +        Steps:
> +            * Launch testpmd session with a physical NIC and virtio-user
> vdev
> +                connected to a vhost-net socket.
> +            * Configure the tap interface that is created with IP address
> and
> +                set link state to UP.
> +            * Launch second testpmd session with af_packet vdev connected
> to
> +                the tap interface.
> +            * Start packet forwarding on both testpmd sessions.
> +            * Send 100 packets to the physical interface from external
> tester.
> +            * Capture packets on the same physical interface.
> +
> +        Verify:
> +            * Physical interface receives all 100 sent packets.
> +        """
> +        self.sut_node = self._ctx.sut_node
> +        if not isinstance(self.sut_node.main_session, LinuxSession):
> +            self.verify(False, "Must be running on a Linux environment.")
> +        with TestPmd(
> +            prefix="virtio",
> +
> vdevs=[VirtualDevice("virtio_user0,path=/dev/vhost-net,queues=1,queue_size=1024")],
> +        ) as virtio:
> +            self.sut_node.main_session.send_command("ip link set dev tap0
> up", privileged=True)
> +            with TestPmd(
> +                prefix="vhost", no_pci=True,
> vdevs=[VirtualDevice("net_af_packet0,iface=tap0")]
> +            ) as vhost:
> +                virtio.set_forward_mode(SimpleForwardingModes.mac)
> +                vhost.set_forward_mode(SimpleForwardingModes.mac)
> +                vhost.start()
> +                virtio.start()
> +
> +                packet = Ether() / IP()
> +                packets = [packet] * 100
> +                captured_packets = self.send_packets_and_capture(packets)
> +
> +                self.verify(
> +                    len(captured_packets) >= 100, "Sent packets not
> received on physical interface."
> +                )
>

Can you remind me what docs you were working from for setting up the pvp
test? I'm guessing how you've done it is valid, but I originally thought we
needed 2 tap interfaces and that vhost had to run in --no-pci, not virtio.
Better safe than sorry. :)


> --
> 2.51.0
>
>
Reviewed-by: Patrick Robb <probb@iol.unh.edu>