Skip to content

Commit

Permalink
q-dev: update tests
Browse files Browse the repository at this point in the history
piotrbartman committed Jun 12, 2024

Verified

This commit was signed with the committer’s verified signature. The key has expired.
piotrbartman Piotr Bartman-Szwarc
1 parent d430f7f commit eae86a6
Showing 4 changed files with 287 additions and 132 deletions.
8 changes: 1 addition & 7 deletions qubes/devices.py
Original file line number Diff line number Diff line change
@@ -68,12 +68,6 @@
from qubes.api import ProtocolError


class DeviceNotAttached(qubes.exc.QubesException, KeyError):
"""
Trying to detach not attached device.
"""


class DeviceNotAssigned(qubes.exc.QubesException, KeyError):
"""
Trying to unassign not assigned device.
@@ -1115,7 +1109,7 @@ async def update_assignment(
await self._vm.fire_event_async(
'device-assignment-changed:' + self._bus, device=device)
else:
await self.detach(assignment)
await self.unassign(assignment)

async def detach(self, device: Device):
"""
2 changes: 1 addition & 1 deletion qubes/tests/api_admin.py
Original file line number Diff line number Diff line change
@@ -1910,7 +1910,7 @@ def test_491_vm_device_detach_not_attached(self):
self.vm.add_handler('device-detach:testclass', mock_detach)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
with self.assertRaises(qubes.devices.DeviceNotAttached):
with self.assertRaises(qubes.devices.DeviceNotAssigned):
self.call_mgmt_func(b'admin.vm.device.testclass.Detach',
b'test-vm1', b'test-vm1+1234')
self.assertFalse(mock_detach.called)
344 changes: 258 additions & 86 deletions qubes/tests/devices.py
Original file line number Diff line number Diff line change
@@ -21,10 +21,11 @@
#

import qubes.devices
from qubes.devices import DeviceInfo, DeviceCategory
from qubes.devices import DeviceInfo

import qubes.tests


class TestDevice(qubes.devices.DeviceInfo):
# pylint: disable=too-few-public-methods
pass
@@ -46,7 +47,7 @@ def __init__(self, app, name, *args, **kwargs):
super(TestVM, self).__init__(*args, **kwargs)
self.app = app
self.name = name
self.device = TestDevice(self, 'testdev', 'Description')
self.device = TestDevice(self, 'testdev', 'testclass')
self.events_enabled = True
self.devices = {
'testclass': qubes.devices.DeviceCollection(self, 'testclass')
@@ -59,7 +60,7 @@ def __str__(self):
return self.name

@qubes.events.handler('device-list-attached:testclass')
def dev_testclass_list_attached(self, event, persistent = False):
def dev_testclass_list_attached(self, event, persistent=False):
for vm in self.app.domains:
if vm.device.data.get('test_frontend_domain', None) == self:
yield (vm.device, {})
@@ -95,115 +96,252 @@ def setUp(self):
required=True,
)

def attach(self):
self.emitter.running = True
# device-attach event not implemented, so manipulate object manually
self.device.data['test_frontend_domain'] = self.emitter

def detach(self):
# device-detach event not implemented, so manipulate object manually
del self.device.data['test_frontend_domain']

def test_000_init(self):
self.assertFalse(self.collection._set)

def test_001_attach(self):
self.emitter.running = True
self.loop.run_until_complete(self.collection.attach(self.assignment))
self.assertEventFired(self.emitter, 'device-pre-attach:testclass')
self.assertEventFired(self.emitter, 'device-attach:testclass')
self.assertEventNotFired(self.emitter, 'device-pre-detach:testclass')
self.assertEventNotFired(self.emitter, 'device-detach:testclass')

def test_002_detach(self):
self.loop.run_until_complete(self.collection.attach(self.assignment))
def test_002_attach_to_halted(self):
with self.assertRaises(qubes.exc.QubesVMNotRunningError):
self.loop.run_until_complete(
self.collection.attach(self.assignment))

def test_003_detach(self):
self.attach()
self.loop.run_until_complete(self.collection.detach(self.assignment))
self.assertEventFired(self.emitter, 'device-pre-attach:testclass')
self.assertEventFired(self.emitter, 'device-attach:testclass')
self.assertEventFired(self.emitter, 'device-pre-detach:testclass')
self.assertEventFired(self.emitter, 'device-detach:testclass')

def test_004_detach_from_halted(self):
with self.assertRaises(LookupError):
self.loop.run_until_complete(
self.collection.detach(self.assignment))

def test_010_empty_detach(self):
self.emitter.running = True
with self.assertRaises(LookupError):
self.loop.run_until_complete(
self.collection.detach(self.assignment))

def test_011_double_attach(self):
self.loop.run_until_complete(self.collection.attach(self.assignment))
def test_011_empty_unassign(self):
for _ in range(2):
with self.assertRaises(LookupError):
self.loop.run_until_complete(
self.collection.unassign(self.assignment))
self.emitter.running = True

def test_012_double_attach(self):
self.attach()
with self.assertRaises(qubes.devices.DeviceAlreadyAttached):
self.loop.run_until_complete(
self.collection.attach(self.assignment))

def test_012_double_detach(self):
self.loop.run_until_complete(self.collection.attach(self.assignment))
def test_013_double_detach(self):
self.attach()
self.loop.run_until_complete(self.collection.detach(self.assignment))
self.detach()

with self.assertRaises(qubes.devices.DeviceNotAttached):
with self.assertRaises(qubes.devices.DeviceNotAssigned):
self.loop.run_until_complete(
self.collection.detach(self.assignment))

def test_013_list_attached_persistent(self):
def test_014_double_assign(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))

with self.assertRaises(qubes.devices.DeviceAlreadyAssigned):
self.loop.run_until_complete(
self.collection.assign(self.assignment))

def test_015_double_unassign(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.loop.run_until_complete(self.collection.unassign(self.assignment))

with self.assertRaises(qubes.devices.DeviceNotAssigned):
self.loop.run_until_complete(
self.collection.unassign(self.assignment))

def test_016_list_assigned(self):
self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.attach(self.assignment))
self.assertEventFired(self.emitter, 'device-list-attached:testclass')
self.assertEqual({self.device}, set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEqual({self.device},
set(self.collection.get_assigned_devices()))
self.assertEqual(set([]),
set(self.collection.get_attached_devices()))
self.assertEqual({self.device},
set(self.collection.get_dedicated_devices()))

def test_014_list_attached_non_persistent(self):
self.assignment.persistent = False
self.emitter.running = True
self.loop.run_until_complete(self.collection.attach(self.assignment))
# device-attach event not implemented, so manipulate object manually
self.device.data['test_frontend_domain'] = self.emitter
def test_017_list_attached(self):
self.assignment.required = False
self.attach()
self.assertEqual({self.device},
set(self.collection.get_attached_devices()))
self.assertEqual(set([]),
set(self.collection.get_assigned_devices()))
self.assertEqual({self.device},
set(self.collection.get_attached_devices()))
set(self.collection.get_dedicated_devices()))
self.assertEventFired(self.emitter, 'device-list-attached:testclass')

def test_015_list_available(self):
def test_018_list_available(self):
self.assertEqual({self.device}, set(self.collection))
self.assertEventFired(self.emitter, 'device-list:testclass')

def test_020_update_persistent_to_false(self):
self.emitter.running = True
def test_020_update_required_to_false(self):
self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.attach(self.assignment))
# device-attach event not implemented, so manipulate object manually
self.device.data['test_frontend_domain'] = self.emitter
self.assertEqual({self.device}, set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.assertEqual({self.device}, set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.collection.update_assignment(self.device, False)
self.assertEqual(set(), set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.attach()
self.assertEqual(
{self.device},
set(self.collection.get_assigned_devices(required_only=True)))
self.assertEqual(
{self.device}, set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(
self.collection.update_assignment(self.device, False))
self.assertEqual(
{self.device}, set(self.collection.get_assigned_devices()))
self.assertEqual(
{self.device}, set(self.collection.get_attached_devices()))

def test_021_update_persistent_to_true(self):
self.assignment.persistent = False
self.emitter.running = True
def test_021_update_required_to_none(self):
self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.attach(self.assignment))
# device-attach event not implemented, so manipulate object manually
self.device.data['test_frontend_domain'] = self.emitter
self.assertEqual(set(), set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.attach()
self.assertEqual(
set(),
set(self.collection.get_assigned_devices(required_only=True)))
self.assertEqual(
{self.device}, set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(
self.collection.update_assignment(self.device, None))
self.assertEqual(
set(), set(self.collection.get_assigned_devices()))
self.assertEqual(
{self.device}, set(self.collection.get_attached_devices()))

def test_022_update_required_to_true(self):
self.assignment.required = False
self.attach()
self.assertEqual(set(), set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.collection.update_assignment(self.device, True)
self.assertEqual({self.device}, set(self.collection.get_assigned_devices()))
self.assertEqual({self.device}, set(self.collection.get_attached_devices()))
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEqual(
set(),
set(self.collection.get_assigned_devices(required_only=True)))
self.assertEqual({self.device},
set(self.collection.get_attached_devices()))
self.assertEqual({self.device}
, set(self.collection.get_assigned_devices()))
self.assertEqual({self.device},
set(self.collection.get_attached_devices()))
self.loop.run_until_complete(
self.collection.update_assignment(self.device, True))
self.assertEqual({self.device},
set(self.collection.get_assigned_devices()))
self.assertEqual({self.device},
set(self.collection.get_attached_devices()))

def test_022_update_persistent_reject_not_running(self):
def test_023_update_required_reject_not_running(self):
self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.attach(self.assignment))
self.assertEqual({self.device}, set(self.collection.get_assigned_devices()))
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEqual({self.device},
set(self.collection.get_assigned_devices()))
self.assertEqual(set(), set(self.collection.get_attached_devices()))
with self.assertRaises(qubes.exc.QubesVMNotStartedError):
self.collection.update_assignment(self.device, False)
self.loop.run_until_complete(
self.collection.update_assignment(self.device, False))

def test_023_update_persistent_reject_not_attached(self):
def test_024_update_required_reject_not_attached(self):
self.assertEqual(set(), set(self.collection.get_assigned_devices()))
self.assertEqual(set(), set(self.collection.get_attached_devices()))
self.emitter.running = True
with self.assertRaises(qubes.exc.QubesValueError):
self.collection.update_assignment(self.device, True)
self.loop.run_until_complete(
self.collection.update_assignment(self.device, True))
with self.assertRaises(qubes.exc.QubesValueError):
self.collection.update_assignment(self.device, False)
self.loop.run_until_complete(
self.collection.update_assignment(self.device, False))

def test_030_assign(self):
self.emitter.running = True
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventNotFired(self.emitter, 'device-unassign:testclass')

def test_031_assign_to_halted(self):
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventNotFired(self.emitter, 'device-unassign:testclass')

def test_032_assign_required(self):
self.emitter.running = True
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventNotFired(self.emitter, 'device-unassign:testclass')

def test_033_assign_required_to_halted(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventNotFired(self.emitter, 'device-unassign:testclass')

def test_034_unassign_from_halted(self):
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.loop.run_until_complete(self.collection.unassign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventFired(self.emitter, 'device-unassign:testclass')

def test_035_unassign(self):
self.emitter.running = True
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.loop.run_until_complete(self.collection.unassign(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventFired(self.emitter, 'device-unassign:testclass')

def test_040_detach_required(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.attach()
with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
self.loop.run_until_complete(
self.collection.detach(self.assignment))

def test_041_detach_required_from_halted(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))
with self.assertRaises(LookupError):
self.loop.run_until_complete(
self.collection.detach(self.assignment))

def test_042_unassign_required(self):
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.emitter.running = True
with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
self.loop.run_until_complete(
self.collection.unassign(self.assignment))

def test_043_detach_assigned(self):
self.assignment.required = False
self.loop.run_until_complete(self.collection.assign(self.assignment))
self.attach()
self.loop.run_until_complete(self.collection.detach(self.assignment))
self.assertEventFired(self.emitter, 'device-assign:testclass')
self.assertEventFired(self.emitter, 'device-pre-detach:testclass')
self.assertEventFired(self.emitter, 'device-detach:testclass')


class TC_01_DeviceManager(qubes.tests.QubesTestCase):
@@ -223,8 +361,9 @@ def test_001_missing(self):
ident=device.ident,
attach_automatically=True, required=True)
self.loop.run_until_complete(
self.manager['testclass'].attach(assignment))
self.assertEventFired(self.emitter, 'device-attach:testclass')
self.manager['testclass'].assign(assignment))
self.assertEqual(
len(list(self.manager['testclass'].get_assigned_devices())), 1)


class TC_02_DeviceInfo(qubes.tests.QubesTestCase):
@@ -233,9 +372,6 @@ def setUp(self):
self.app = TestApp()
self.vm = TestVM(self.app, 'vm')

def test_001_init(self):
pass # TODO

def test_010_serialize(self):
device = DeviceInfo(
backend_domain=self.vm,
@@ -246,41 +382,77 @@ def test_010_serialize(self):
manufacturer="",
name="Some untrusted garbage",
serial=None,
interfaces=[DeviceCategory.Other, DeviceCategory.USB_HID],
# additional_info="", # TODO
# date="06.12.23", # TODO
interfaces=[qubes.devices.DeviceInterface(" ******"),
qubes.devices.DeviceInterface("u03**01")],
additional_info="",
date="06.12.23",
)
actual = device.serialize()
expected = (
b"manufacturer='unknown' self_identity='0000:0000::?******' "
b"serial='unknown' ident='1-1.1.1' product='Qubes' "
b"vendor='ITL' name='Some untrusted garbage' devclass='bus' "
b"backend_domain='vm' interfaces=' ******u03**01' "
b"_additional_info='' _date='06.12.23'")
expected = set(expected.replace(b"Some untrusted garbage",
b"Some_untrusted_garbage").split(b" "))
actual = set(actual.replace(b"Some untrusted garbage",
b"Some_untrusted_garbage").split(b" "))
self.assertEqual(actual, expected)

def test_011_serialize_with_parent(self):
device = DeviceInfo(
backend_domain=self.vm,
ident="1-1.1.1",
devclass="bus",
vendor="ITL",
product="Qubes",
manufacturer="",
name="Some untrusted garbage",
serial=None,
interfaces=[qubes.devices.DeviceInterface(" ******"),
qubes.devices.DeviceInterface("u03**01")],
additional_info="",
date="06.12.23",
parent=qubes.devices.Device(self.vm, '1-1.1', 'pci')
)
actual = sorted(device.serialize().split(b' '))
expected = [
b'YmFja2VuZF9kb21haW49dm0=', b'ZGV2Y2xhc3M9YnVz',
b'aW50ZXJmYWNlcz0qKioqKiowMyoqKio=', b'aWRlbnQ9MS0xLjEuMQ==',
b'bWFudWZhY3R1cmVyPXVua25vd24=',
b'bmFtZT1Tb21lIHVudHJ1c3RlZCBnYXJiYWdl',
b'c2VyaWFsPXVua25vd24=', b'cHJvZHVjdD1RdWJlcw==',
b'dmVuZG9yPUlUTA==']
actual = device.serialize()
expected = (
b"manufacturer='unknown' self_identity='0000:0000::?******' "
b"serial='unknown' ident='1-1.1.1' product='Qubes' "
b"vendor='ITL' name='Some untrusted garbage' devclass='bus' "
b"backend_domain='vm' interfaces=' ******u03**01' "
b"_additional_info='' _date='06.12.23' "
b"parent_ident='1-1.1' parent_devclass='pci'")
expected = set(expected.replace(b"Some untrusted garbage",
b"Some_untrusted_garbage").split(b" "))
actual = set(actual.replace(b"Some untrusted garbage",
b"Some_untrusted_garbage").split(b" "))
self.assertEqual(actual, expected)

def test_020_deserialize(self):
expected = [
b'YmFja2VuZF9kb21haW49dm0=', b'ZGV2Y2xhc3M9YnVz',
b'aW50ZXJmYWNlcz0qKioqKiowMyoqKio=', b'aWRlbnQ9MS0xLjEuMQ==',
b'bWFudWZhY3R1cmVyPXVua25vd24=',
b'bmFtZT1Tb21lIHVudHJ1c3RlZCBnYXJiYWdl',
b'c2VyaWFsPXVua25vd24=', b'cHJvZHVjdD1RdWJlcw==',
b'dmVuZG9yPUlUTA==']
actual = DeviceInfo.deserialize(b' '.join(expected), self.vm)
serialized = (
b"1-1.1.1 "
b"manufacturer='unknown' self_identity='0000:0000::?******' "
b"serial='unknown' ident='1-1.1.1' product='Qubes' "
b"vendor='ITL' name='Some untrusted garbage' devclass='bus' "
b"backend_domain='vm' interfaces=' ******u03**01' "
b"_additional_info='' _date='06.12.23' "
b"parent_ident='1-1.1' parent_devclass='None'")
actual = DeviceInfo.deserialize(serialized, self.vm)
expected = DeviceInfo(
backend_domain=self.vm,
ident="1-1.1.1",
devclass="bus",
vendor="ITL",
product="Qubes",
manufacturer="",
manufacturer="unknown",
name="Some untrusted garbage",
serial=None,
interfaces=[DeviceCategory.Other, DeviceCategory.USB_HID],
# additional_info="", # TODO
# date="06.12.23", # TODO
interfaces=[qubes.devices.DeviceInterface(" ******"),
qubes.devices.DeviceInterface("u03**01")],
additional_info="",
date="06.12.23",
)

self.assertEqual(actual.backend_domain, expected.backend_domain)
@@ -291,6 +463,6 @@ def test_020_deserialize(self):
self.assertEqual(actual.manufacturer, expected.manufacturer)
self.assertEqual(actual.name, expected.name)
self.assertEqual(actual.serial, expected.serial)
self.assertEqual(actual.interfaces, expected.interfaces)
# self.assertEqual(actual.data, expected.data) # TODO

self.assertEqual(repr(actual.interfaces), repr(expected.interfaces))
self.assertEqual(actual.self_identity, expected.self_identity)
self.assertEqual(actual.data, expected.data)
65 changes: 27 additions & 38 deletions qubes/tests/devices_block.py
Original file line number Diff line number Diff line change
@@ -115,8 +115,11 @@ def __init__(self):
self.domains = {}


class TestVM(object):
def __init__(self, qdb, domain_xml=None, running=True, name='test-vm'):
class TestVM(qubes.tests.TestEmitter):
def __init__(
self, qdb, domain_xml=None, running=True, name='test-vm',
*args, **kwargs):
super(TestVM, self).__init__(*args, **kwargs)
self.name = name
self.untrusted_qdb = TestQubesDB(qdb)
self.libvirt_domain = mock.Mock()
@@ -132,6 +135,9 @@ def __init__(self, qdb, domain_xml=None, running=True, name='test-vm'):
self.libvirt_domain.configure_mock(**{
'XMLDesc.return_value': domain_xml
})
self.devices = {
'testclass': qubes.devices.DeviceCollection(self, 'testclass')
}

def __eq__(self, other):
if isinstance(other, TestVM):
@@ -147,16 +153,18 @@ def setUp(self):
def test_000_device_get(self):
vm = TestVM({
'/qubes-block-devices/sda': b'',
'/qubes-block-devices/sda/desc': b'Test device',
'/qubes-block-devices/sda/desc': b'Test_ (device)',
'/qubes-block-devices/sda/size': b'1024000',
'/qubes-block-devices/sda/mode': b'w',
})
device_info = self.ext.device_get(vm, 'sda')
self.assertIsInstance(device_info, qubes.ext.block.BlockDevice)
self.assertEqual(device_info.backend_domain, vm)
self.assertEqual(device_info.ident, 'sda')
self.assertEqual(device_info.name, 'Test device')
self.assertEqual(device_info._name, 'Test device')
self.assertEqual(device_info.name, 'device')
self.assertEqual(device_info._name, 'device')
self.assertEqual(device_info.serial, 'Test')
self.assertEqual(device_info._serial, 'Test')
self.assertEqual(device_info.size, 1024000)
self.assertEqual(device_info.mode, 'w')
self.assertEqual(
@@ -166,16 +174,18 @@ def test_000_device_get(self):
def test_001_device_get_other_node(self):
vm = TestVM({
'/qubes-block-devices/mapper_dmroot': b'',
'/qubes-block-devices/mapper_dmroot/desc': b'Test device',
'/qubes-block-devices/mapper_dmroot/desc': b'Test_device',
'/qubes-block-devices/mapper_dmroot/size': b'1024000',
'/qubes-block-devices/mapper_dmroot/mode': b'w',
})
device_info = self.ext.device_get(vm, 'mapper_dmroot')
self.assertIsInstance(device_info, qubes.ext.block.BlockDevice)
self.assertEqual(device_info.backend_domain, vm)
self.assertEqual(device_info.ident, 'mapper_dmroot')
self.assertEqual(device_info.name, 'Test device')
self.assertEqual(device_info._name, 'Test device')
self.assertEqual(device_info._name, None)
self.assertEqual(device_info.name, 'unknown')
self.assertEqual(device_info.serial, 'Test device')
self.assertEqual(device_info._serial, 'Test device')
self.assertEqual(device_info.size, 1024000)
self.assertEqual(device_info.mode, 'w')
self.assertEqual(
@@ -185,12 +195,13 @@ def test_001_device_get_other_node(self):
def test_002_device_get_invalid_desc(self):
vm = TestVM({
'/qubes-block-devices/sda': b'',
'/qubes-block-devices/sda/desc': b'Test device<>za\xc4\x87abc',
'/qubes-block-devices/sda/desc': b'Test (device<>za\xc4\x87abc)',
'/qubes-block-devices/sda/size': b'1024000',
'/qubes-block-devices/sda/mode': b'w',
})
device_info = self.ext.device_get(vm, 'sda')
self.assertEqual(device_info.name, 'Test device__za__abc')
self.assertEqual(device_info.serial, 'Test')
self.assertEqual(device_info.name, 'device zaabc')

def test_003_device_get_invalid_size(self):
vm = TestVM({
@@ -227,24 +238,26 @@ def test_005_device_get_none(self):
def test_010_devices_list(self):
vm = TestVM({
'/qubes-block-devices/sda': b'',
'/qubes-block-devices/sda/desc': b'Test device',
'/qubes-block-devices/sda/desc': b'Test_device',
'/qubes-block-devices/sda/size': b'1024000',
'/qubes-block-devices/sda/mode': b'w',
'/qubes-block-devices/sdb': b'',
'/qubes-block-devices/sdb/desc': b'Test device2',
'/qubes-block-devices/sdb/desc': b'Test_device (2)',
'/qubes-block-devices/sdb/size': b'2048000',
'/qubes-block-devices/sdb/mode': b'r',
})
devices = sorted(list(self.ext.on_device_list_block(vm, '')))
self.assertEqual(len(devices), 2)
self.assertEqual(devices[0].backend_domain, vm)
self.assertEqual(devices[0].ident, 'sda')
self.assertEqual(devices[0].description, 'Test device')
self.assertEqual(devices[0].serial, 'Test device')
self.assertEqual(devices[0].name, 'unknown')
self.assertEqual(devices[0].size, 1024000)
self.assertEqual(devices[0].mode, 'w')
self.assertEqual(devices[1].backend_domain, vm)
self.assertEqual(devices[1].ident, 'sdb')
self.assertEqual(devices[1].description, 'Test device2')
self.assertEqual(devices[1].serial, 'Test device')
self.assertEqual(devices[1].name, '2')
self.assertEqual(devices[1].size, 2048000)
self.assertEqual(devices[1].mode, 'r')

@@ -585,27 +598,3 @@ def test_051_detach_not_attached(self):
dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
self.ext.on_device_pre_detached_block(vm, '', dev)
self.assertFalse(vm.libvirt_domain.detachDevice.called)

def test_060_devices_added(self):
vm = TestVM({
'/qubes-block-devices/sda': b'',
'/qubes-block-devices/sda/desc': b'Test device',
'/qubes-block-devices/sda/size': b'1024000',
'/qubes-block-devices/sda/mode': b'w',
'/qubes-block-devices/sdb': b'',
'/qubes-block-devices/sdb/desc': b'Test device2',
'/qubes-block-devices/sdb/size': b'2048000',
'/qubes-block-devices/sdb/mode': b'r',
})
devices = sorted(list(self.ext.on_qdb_change(vm, '')))
self.assertEqual(len(devices), 2)
self.assertEqual(devices[0].backend_domain, vm)
self.assertEqual(devices[0].ident, 'sda')
self.assertEqual(devices[0].description, 'Test device')
self.assertEqual(devices[0].size, 1024000)
self.assertEqual(devices[0].mode, 'w')
self.assertEqual(devices[1].backend_domain, vm)
self.assertEqual(devices[1].ident, 'sdb')
self.assertEqual(devices[1].description, 'Test device2')
self.assertEqual(devices[1].size, 2048000)
self.assertEqual(devices[1].mode, 'r')

0 comments on commit eae86a6

Please sign in to comment.