From 853bcb7574415b850e5fa2c90c2464b63bc6611a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=AE=8B=E4=B8=80=E9=B8=A3?= <1000000008@qq.com>
Date: Tue, 3 Nov 2026 17:50:29 +0800
Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E8=A6=86=E7=9B=96US?=
 =?UTF-8?q?B=E8=AE=BE=E5=A4=87=E6=8C=82=E8=BD=BD=E4=B8=8E=E5=A4=B1?=
 =?UTF-8?q?=E8=B4=A5=E5=9B=9E=E6=BB=9A?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/tests/test_usb_device_qmp.py | 50 ++++++++++++++++++++++++++++++
 1 file changed, 50 insertions(+)
 create mode 100644 tools/tests/test_usb_device_qmp.py

diff --git a/tools/tests/test_usb_device_qmp.py b/tools/tests/test_usb_device_qmp.py
new file mode 100644
index 0000000..a524343
--- /dev/null
+++ b/tools/tests/test_usb_device_qmp.py
@@ -0,0 +1,50 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+spec = importlib.util.spec_from_file_location("usb_qmp", Path(__file__).parents[1] / "usb-device-qmp.py")
+usb = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(usb)
+
+
+class FakeQmp:
+    def __init__(self, occupied=False, fail_add=False):
+        self.calls = []
+        self.occupied = occupied
+        self.fail_add = fail_add
+
+    def call(self, name, args=None):
+        self.calls.append((name, args))
+        if name == "query-chardev":
+            return [{"label": "vdi_usb_box_100"}] if self.occupied else []
+        if name == "qom-list":
+            return [{"name": "tablet"}, {"name": "hostpci0"}]
+        if name == "device_add" and self.fail_add:
+            raise RuntimeError("bad USB bus")
+        return {}
+
+
+class UsbQmpTests(unittest.TestCase):
+    def test_only_loopback_tunnel_and_approved_storage_are_attached(self):
+        q = FakeQmp()
+        usb.attach(q, 100, 40001, "ehci-2.0", "0951", "1666")
+        channel = next(a for n, a in q.calls if n == "chardev-add")
+        self.assertEqual(channel["backend"]["data"]["addr"]["data"]["host"], "127.0.0.1")
+        device = next(a for n, a in q.calls if n == "device_add")
+        self.assertEqual(device["filter"], "0x08:0x0951:0x1666:-1:1|-1:-1:-1:-1:0")
+
+    def test_busy_endpoint_is_never_replaced(self):
+        q = FakeQmp(occupied=True)
+        with self.assertRaisesRegex(RuntimeError, "already exists"):
+            usb.attach(q, 100, 40001, "ehci-2.0", "0951", "1666")
+        self.assertFalse(any(n in ("chardev-add", "chardev-remove", "device_del") for n, _ in q.calls))
+
+    def test_failed_device_add_rolls_back_only_its_new_channel(self):
+        q = FakeQmp(fail_add=True)
+        with self.assertRaisesRegex(RuntimeError, "bad USB bus"):
+            usb.attach(q, 100, 40001, "ehci-2.0", "0951", "1666")
+        self.assertEqual(q.calls[-1], ("chardev-remove", {"id": "vdi_usb_box_100"}))
+
+
+if __name__ == "__main__":
+    unittest.main()