q-dev: fix tests and make linter happy

Author: piotrbartman

qubes/api/admin.py

@@ -27,7 +27,6 @@
 import string
 import subprocess
 import pathlib
-import sys
 
 import libvirt
 import lxml.etree
@@ -46,8 +45,8 @@
 import qubes.vm
 import qubes.vm.adminvm
 import qubes.vm.qubesvm
-from qubes.device_protocol import (Port, VirtualDevice, UnknownDevice,
-                                   DeviceAssignment)
+from qubes.device_protocol import (
+    VirtualDevice, UnknownDevice, DeviceAssignment)
 
 
 class QubesMgmtEventsDispatcher:

qubes/device_protocol.py

@@ -1191,8 +1191,7 @@ def device(self) -> DeviceInfo:
             return devices[0]
         if len(devices) > 1:
             raise ProtocolError("Too many devices matches to assignment")
-        if len(devices) == 0:
-            raise ProtocolError("Any devices matches to assignment")
+        raise ProtocolError("Any devices matches to assignment")
 
     @property
     def port(self) -> Port:

qubes/ext/admin.py

@@ -17,7 +17,6 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 import importlib
-import sys
 
 import qubes.api
 import qubes.api.internal
@@ -171,22 +170,23 @@ def on_tag_add(self, vm, event, tag, **kwargs):
     def on_device_attach(
             self, vm, event, dest, arg, device, mode, options, **kwargs
     ):
+        # pylint: disable=unused-argument
         # ignore auto-attachment
         if mode != 'manual':
             return
 
         # load device deny list
         deny = {}
         try:
-            with open(DEVICE_DENY_LIST, 'r') as file:
+            with open(DEVICE_DENY_LIST, 'r', encoding="utf-8") as file:
                 for line in file:
                     line = line.strip()
 
                     if line:
                         name, *values = line.split()
 
                         values = ' '.join(values).replace(',', ' ').split()
-                        values = set([v for v in values if len(v) > 0])
+                        values = {v for v in values if len(v) > 0}
 
                         deny[name] = deny.get(name, set()).union(set(values))
         except IOError:
@@ -198,4 +198,3 @@ def on_device_attach(
             for devint in device.interfaces:
                 if pattern.matches(devint):
                     raise qubes.exc.PermissionDenied()
-

qubes/ext/pci.py

@@ -223,6 +223,10 @@ def interfaces(self) -> List[qubes.device_protocol.DeviceInterface]:
         Every device should have at least one interface.
         """
         if self._interfaces is None:
+            if self.backend_domain.app.vmm.offline_mode:
+                # don't cache this value
+                return [qubes.device_protocol.DeviceInterface(
+                    '******', devclass='pci')]
             hostdev_details = \
                 self.backend_domain.app.vmm.libvirt_conn.nodeDeviceLookupByName(
                     self.libvirt_name
@@ -287,7 +291,9 @@ def _load_desc(self) -> Dict[str, str]:
                   "manufacturer": unknown,
                   "name": unknown,
                   "serial": unknown}
-        if not self.backend_domain.is_running():
+        if (not self.backend_domain.is_running()
+            or self.backend_domain.app.vmm.offline_mode
+        ):
             # don't cache these values
             return result
         hostdev_details = \

qubes/ext/utils.py

@@ -141,17 +141,16 @@ def compare_device_cache(vm, devices_cache, current_devices):
 
 
 def confirm_device_attachment(device, frontends) -> str:
-    guivm = 'dom0'  # TODO
-
     try:
+        # pylint: disable=consider-using-with
         proc = subprocess.Popen(
-            ["attach-confirm", guivm,
-             device.backend_domain.name, device.port_id,
-             device.description,
+            ["attach-confirm", device.backend_domain.name,
+            device.port_id, device.description,
              *[f.name for f in frontends.keys()]],
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
         (target_name, _) = proc.communicate()
         return target_name.decode()
     except Exception as exc:
-        print(exc, file=sys.stderr)
+        print("attach-confirm", exc, file=sys.stderr)
         return ""

qubes/tests/devices_pci.py

@@ -126,6 +126,7 @@ def setUp(self):
     def test_000_unsupported_device(self):
         vm = TestVM()
         vm.app.configure_mock(**{
+            'vmm.offline_mode': False,
             'vmm.libvirt_conn.nodeDeviceLookupByName.return_value':
                 mock.Mock(**{"XMLDesc.return_value":
                                  PCI_XML.format(*["0000"] * 3)

qubes/tests/vm/init.py

@@ -1,4 +1,5 @@
 # pylint: disable=protected-access
+import sys
 
 #
 # The Qubes OS Project, https://www.qubes-os.org/
@@ -48,6 +49,10 @@ class TestVM(qubes.vm.BaseVM):
     testlabel = qubes.property('testlabel')
     defaultprop = qubes.property('defaultprop', default='defaultvalue')
 
+    def is_running(self):
+        return False
+
+
 class TC_10_BaseVM(qubes.tests.QubesTestCase):
     def setUp(self):
         super().setUp()
@@ -110,8 +115,10 @@ def test_000_load(self):
         })
 
         self.assertCountEqual(vm.devices.keys(), ('pci',))
-        self.assertCountEqual(list(vm.devices['pci'].get_assigned_devices()),
-                              [qubes.ext.pci.PCIDevice(vm, '00_11.22')])
+
+        self.assertTrue(
+            list(vm.devices['pci'].get_assigned_devices())[0].matches(
+                qubes.ext.pci.PCIDevice(vm, '00_11.22', )))
 
         assignments = list(vm.devices['pci'].get_assigned_devices())
         self.assertEqual(len(assignments), 1)

qubes/tests/vm/qubesvm.py

@@ -1893,23 +1893,37 @@ def test_615_libvirt_xml_block_devices(self):
                 'options': {'frontend-dev': 'xvdl'},
                 'device.device_node': '/dev/sdb',
                 'device.backend_domain.name': 'dom0',
+                'devices': [unittest.mock.Mock(**{
+                    'device_node': '/dev/sdb',
+                    'backend_domain.name': 'dom0',})]
             }),
             unittest.mock.Mock(**{
                 'options': {'devtype': 'cdrom'},
                 'device.device_node': '/dev/sda',
                 'device.backend_domain.name': 'dom0',
+                'devices': [unittest.mock.Mock(**{
+                    'device_node': '/dev/sda',
+                    'backend_domain.name': 'dom0', })]
             }),
             unittest.mock.Mock(**{
                 'options': {'read-only': True},
                 'device.device_node': '/dev/loop0',
                 'device.backend_domain.name': 'backend0',
                 'device.backend_domain.features.check_with_template.return_value': '4.2',
+                'devices': [unittest.mock.Mock(**{
+                    'device_node': '/dev/loop0',
+                    'backend_domain.name': 'backend0',
+                    'backend_domain.features.check_with_template.return_value': '4.2'})]
             }),
             unittest.mock.Mock(**{
                 'options': {},
                 'device.device_node': '/dev/loop0',
                 'device.backend_domain.name': 'backend1',
                 'device.backend_domain.features.check_with_template.return_value': '4.2',
+                'devices': [unittest.mock.Mock(**{
+                    'device_node': '/dev/loop0',
+                    'backend_domain.name': 'backend1',
+                    'backend_domain.features.check_with_template.return_value': '4.2'})]
             }),
         ]
         vm.devices['block'].get_assigned_devices = \

qubes/vm/qubesvm.py

@@ -1182,10 +1182,9 @@ async def start(self, start_guid=True, notify_function=None,
                         if not ass.required:
                             continue
                         for device in ass.devices:
-                            if isinstance(device,
-                                          qubes.device_protocol.UnknownDevice):
-                                continue
-                            else:
+                            if not isinstance(
+                                device, qubes.device_protocol.UnknownDevice
+                            ):
                                 break
                         else:
                             raise qubes.exc.QubesException(