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