From 26d6d14f619017e05a8f254ba96f52a68d77b38c Mon Sep 17 00:00:00 2001
From: Kasper Revsbech <kare@universal-robots.com>
Date: Thu, 26 Mar 2026 13:21:28 +0100
Subject: [PATCH] labgrid/qemudriver: add USB network device hotplug support

Use the QMPMonitor to add and remove a USB network device from
a running QEMU instance.

Assumes the QEMU instance is already configured with a USB bus,
e.g. `-device qemu-xhci,id=xhci`

Signed-off-by: Kasper Revsbech <kare@universal-robots.com>
---
 labgrid/driver/qemudriver.py | 42 ++++++++++++++++++++++++++++++++++++
 tests/test_qemudriver.py     | 36 +++++++++++++++++++++++++++++++
 2 files changed, 78 insertions(+)

diff --git a/labgrid/driver/qemudriver.py b/labgrid/driver/qemudriver.py
index a9408874a..c1aaad410 100644
--- a/labgrid/driver/qemudriver.py
+++ b/labgrid/driver/qemudriver.py
@@ -337,6 +337,48 @@ def remove_port_forward(self, proto, local_address, local_port, netdev=""):
             {"command-line": " ".join(command)},
         )
 
+    def add_usb_network_device(self, device_id="usb-net-0", netdev="usbnet", bus="xhci.0"):
+        """Hot-plug a USB network device into the running QEMU instance via QMP.
+
+        Creates a ``user`` netdev backend and attaches a ``usb-net`` device to
+        the ``qemu-xhci`` controller.
+        The controller must already be present at QEMU startup
+        (``-device qemu-xhci,id=xhci`` in ``extra_args``).
+
+        Both the netdev and the device are created at runtime so no netdev
+        backend lingers unattached at boot (which would produce a QEMU warning).
+
+        Args:
+            device_id: Unique QMP device id.  Defaults to ``"usb-net-0"``.
+            netdev:    Id for the netdev backend (created on the fly).
+                       Defaults to ``"usbnet"``.
+            bus:       USB controller bus to attach to.  Defaults to ``"xhci.0"``.
+
+        Example::
+            driver.add_usb_network_device()
+            # or with explicit args:
+            driver.add_usb_network_device(device_id="usb-net-0", netdev="usbnet", bus="xhci.0")
+        """
+        self.monitor_command(
+            "netdev_add",
+            {"type": "user", "id": netdev},
+        )
+        self.monitor_command(
+            "device_add",
+            {"driver": "usb-net", "id": device_id, "netdev": netdev, "bus": bus},
+        )
+
+    def remove_usb_network_device(
+        self, device_id="usb-net-0", netdev="usbnet"):
+        """Hot-unplug a USB network device from the running QEMU instance via QMP.
+
+        Args:
+            device_id: QMP device id to remove.  Defaults to ``"usb-net-0"``.
+            netdev:    Id of the netdev backend to destroy.  Defaults to ``"usbnet"``.
+        """
+        self.monitor_command("device_del", {"id": device_id})
+        self.monitor_command("netdev_del", {"id": netdev})
+
     def _read(self, size=1, timeout=10, max_size=None):
         ready, _, _ = select.select([self._clientsocket], [], [], timeout)
         if ready:
diff --git a/tests/test_qemudriver.py b/tests/test_qemudriver.py
index 0b137da2a..748e503e0 100644
--- a/tests/test_qemudriver.py
+++ b/tests/test_qemudriver.py
@@ -112,3 +112,39 @@ def test_qemu_port_forwarding_with_netdev(qemu_target, qemu_driver, qemu_mock, q
     assert qemu_driver._forwarded_ports == {}
 
     qemu_target.deactivate(qemu_driver)
+
+def test_add_usb_network_device(qemu_target, qemu_driver, qemu_mock, qemu_version_mock, mocker):
+    qemu_target.activate(qemu_driver)
+
+    qemu_driver.on()
+    qemu_driver.qmp.execute = mocker.MagicMock(return_value={})
+    qemu_driver.add_usb_network_device()
+    qemu_driver.qmp.execute.assert_any_call("netdev_add", {"type": "user", "id": "usbnet"})
+    qemu_driver.qmp.execute.assert_any_call("device_add", {"driver": "usb-net", "id": "usb-net-0", "netdev": "usbnet", "bus": "xhci.0"})
+
+    qemu_target.deactivate(qemu_driver)
+
+
+def test_add_usb_network_device_custom_args(qemu_target, qemu_driver, qemu_mock, qemu_version_mock, mocker):
+    qemu_target.activate(qemu_driver)
+
+    qemu_driver.on()
+    qemu_driver.qmp.execute = mocker.MagicMock(return_value={})
+    qemu_driver.add_usb_network_device(device_id="usb-net-1", netdev="mynet", bus="xhci.0")
+    qemu_driver.qmp.execute.assert_any_call("netdev_add", {"type": "user", "id": "mynet"})
+    qemu_driver.qmp.execute.assert_any_call("device_add", {"driver": "usb-net", "id": "usb-net-1", "netdev": "mynet", "bus": "xhci.0"})
+
+    qemu_target.deactivate(qemu_driver)
+
+
+def test_remove_usb_network_device(qemu_target, qemu_driver, qemu_mock, qemu_version_mock, mocker):
+    qemu_target.activate(qemu_driver)
+
+    qemu_driver.on()
+    qemu_driver.qmp.execute = mocker.MagicMock(return_value={})
+    qemu_driver.add_usb_network_device()
+    qemu_driver.remove_usb_network_device()
+    qemu_driver.qmp.execute.assert_any_call("device_del", {"id": "usb-net-0"})
+    qemu_driver.qmp.execute.assert_any_call("netdev_del", {"id": "usbnet"})
+
+    qemu_target.deactivate(qemu_driver)