Skip to content

Commit

Permalink
q-dev: keep partial backward compatibility in tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@piotrbartman
piotrbartman committed Jun 10, 2024
1 parent cad623a commit 227ec98
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 36 deletions.
8 changes: 4 additions & 4 deletions qubesusbproxy/core3ext.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ def get_assigned_devices(devices):
class DescriptionOverrider:
@property
def description(self):
return self.vendor + " " + self.product
return self.name

class DeviceInfo(DescriptionOverrider, LegacyDeviceInfo):
def __init__(self, *args, **kwargs):
Expand All @@ -70,9 +70,9 @@ def __init__(self, *args, **kwargs):
# `_load_interfaces_from_qubesdb` will never be called
self._interfaces = "?******"

@property
def fronted_domain(self):
return self.attachment
@property
def frontend_domain(self):
return self.attachment

class DeviceInterface:
pass
Expand Down
78 changes: 46 additions & 32 deletions qubesusbproxy/tests.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,35 @@

core2 = False
core3 = False
legacy = False
try:
import qubesusbproxy.core3ext
import qubes.devices
import asyncio

try:
from qubes.device_protocol import DeviceAssignment

def assign(test, collection, assignment):
test.loop.run_until_complete(collection.assign(assignment))

def unassign(test, collection, assignment):
test.loop.run_until_complete(collection.unassign(assignment))

AUTO_ATTACH = {"attach_automatically": True, "required": True}
except ImportError:
# This extension supports both the legacy and new device API.
# In the case of the legacy backend, functionality is limited.
from qubes.devices import DeviceAssignment

def assign(test, collection, assignment):
test.loop.run_until_complete(collection.attach(assignment))

def unassign(test, collection, assignment):
test.loop.run_until_complete(collection.detach(assignment))

legacy = True
AUTO_ATTACH = {"persistent": True}

core3 = True
except ImportError:
pass
Expand Down Expand Up @@ -380,11 +404,9 @@ def test_000_list(self):

def test_010_assign(self):
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
assign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)
try:
self.frontend.start()
Expand All @@ -395,13 +417,12 @@ def test_010_assign(self):
wait=True), 0,
"Device connection failed")

self.assertEqual(usb_dev.attachment,
self.frontend)
self.assertEqual(usb_dev.attachment, self.frontend)

def test_020_attach(self):
self.frontend.start()
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -412,13 +433,12 @@ def test_020_attach(self):
wait=True), 0,
"Device connection failed")

self.assertEquals(usb_dev.attachment,
self.frontend)
self.assertEquals(usb_dev.attachment, self.frontend)

def test_030_detach(self):
self.frontend.start()
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -438,14 +458,11 @@ def test_030_detach(self):

def test_040_unassign(self):
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
assign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)
self.loop.run_until_complete(
self.frontend.devices['usb'].unassign(ass))
unassign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)

def test_050_list_attached(self):
Expand All @@ -454,7 +471,7 @@ def test_050_list_attached(self):
usb_list = self.backend.devices['usb']

usb_list_front_pre = list(self.frontend.devices['usb'])
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)

try:
self.loop.run_until_complete(
Expand All @@ -466,8 +483,7 @@ def test_050_list_attached(self):
wait=True), 0,
"Device connection failed")

self.assertEquals(usb_list[self.usbdev_ident].attachment,
self.frontend)
self.assertEquals(usb_list[self.usbdev_ident].attachment, self.frontend)

usb_list_front_post = list(self.frontend.devices['usb'])

Expand All @@ -476,7 +492,7 @@ def test_050_list_attached(self):
def test_060_auto_detach_on_remove(self):
self.frontend.start()
usb_list = self.backend.devices['usb']
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -495,12 +511,10 @@ def test_060_auto_detach_on_remove(self):
def test_061_auto_attach_on_reconnect(self):
self.frontend.start()
usb_list = self.backend.devices['usb']
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
assign(self, self.frontend.devices['usb'], ass)
except qubesusbproxy.core3ext.USBProxyNotInstalled as e:
self.skipTest(str(e))

Expand Down Expand Up @@ -530,7 +544,7 @@ def test_070_attach_not_installed_front(self):
user="root", wait=True)
if retcode != 0:
raise RuntimeError("Failed to simulate not installed package")
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
with self.assertRaises(qubesusbproxy.core3ext.USBProxyNotInstalled):
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -543,7 +557,7 @@ def test_075_attach_not_installed_back(self):
user="root", wait=True)
if retcode != 0:
raise RuntimeError("Failed to simulate not installed package")
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
with self.assertRaises(qubesusbproxy.core3ext.USBProxyNotInstalled):
self.loop.run_until_complete(
Expand All @@ -560,7 +574,7 @@ def test_080_attach_existing_policy(self):
'/etc/qubes-rpc/policy/qubes.USB+{}'.format(self.usbdev_ident),
'w+') as policy_file:
policy_file.write('# empty policy\n')
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))

Expand All @@ -569,7 +583,7 @@ def test_090_attach_stubdom(self):
self.frontend.virt_mode = 'hvm'
self.frontend.features['stubdom-qrexec'] = True
self.frontend.start()
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand Down

0 comments on commit 227ec98

Please sign in to comment.