51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
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()
|