labgrid/qemudriver: add USB network device hotplug support

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:

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)