Fix qubes.tests.devices

Signed-off-by: Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>

qubes/tests/devices.py

@@ -51,12 +51,13 @@ class TestVM(qubes.tests.TestEmitter):
         }
         self.app.domains[name] = self
         self.app.domains[self] = self
+        self.running = False
 
     def __str__(self):
         return self.name
 
     @qubes.events.handler('device-list-attached:testclass')
-    def dev_testclass_list_attached(self, event, persistent):
+    def dev_testclass_list_attached(self, event, persistent = False):
         for vm in self.app.domains:
             if vm.device.frontend_domain == self:
                 yield vm.device
@@ -65,6 +66,10 @@ class TestVM(qubes.tests.TestEmitter):
     def dev_testclass_list(self, event):
         yield self.device
 
+    def is_halted(self):
+        return not self.running
+
+
 
 class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
     def setUp(self):
@@ -73,20 +78,25 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
         self.app.domains['vm'] = self.emitter
         self.device = self.emitter.device
         self.collection = self.emitter.devices['testclass']
+        self.assignment = qubes.devices.DeviceAssignment(
+            backend_domain = self.device.backend_domain,
+            ident = self.device.ident,
+            persistent=True
+        )
 
     def test_000_init(self):
         self.assertFalse(self.collection._set)
 
     def test_001_attach(self):
-        self.collection.attach(self.device)
+        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.collection.attach(self.device)
-        self.collection.detach(self.device)
+        self.collection.attach(self.assignment)
+        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')
@@ -94,41 +104,43 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
 
     def test_010_empty_detach(self):
         with self.assertRaises(LookupError):
-            self.collection.detach(self.device)
+            self.collection.detach(self.assignment)
 
     def test_011_double_attach(self):
-        self.collection.attach(self.device)
+        self.collection.attach(self.assignment)
 
         with self.assertRaises(LookupError):
-            self.collection.attach(self.device)
+            self.collection.attach(self.assignment)
 
     def test_012_double_detach(self):
-        self.collection.attach(self.device)
-        self.collection.detach(self.device)
+        self.collection.attach(self.assignment)
+        self.collection.detach(self.assignment)
 
-        with self.assertRaises(LookupError):
-            self.collection.detach(self.device)
+        with self.assertRaises(qubes.devices.DeviceNotAttached):
+            self.collection.detach(self.assignment)
 
     def test_013_list_attached_persistent(self):
-        self.assertEqual(set([]), set(self.collection.attached()))
+        self.assertEqual(set([]), set(self.collection.persistent()))
+        self.collection.attach(self.assignment)
         self.assertEventFired(self.emitter, 'device-list-attached:testclass')
-        self.collection.attach(self.device)
-        self.assertEqual({self.device}, set(self.collection.attached()))
+        self.assertEqual({self.device}, set(self.collection.persistent()))
         self.assertEqual({self.device},
-            set(self.collection.attached(persistent=True)))
+            set(self.collection.persistent()))
         self.assertEqual(set([]),
-            set(self.collection.attached(persistent=False)))
+            set(self.collection.attached()))
 
     def test_014_list_attached_non_persistent(self):
-        self.collection.attach(self.device, persistent=False)
+        self.assignment.persistent = False
+        self.emitter.running = True
+        self.collection.attach(self.assignment)
         # device-attach event not implemented, so manipulate object manually
         self.device.frontend_domain = self.emitter
         self.assertEqual({self.device},
             set(self.collection.attached()))
         self.assertEqual(set([]),
-            set(self.collection.attached(persistent=True)))
+            set(self.collection.persistent()))
         self.assertEqual({self.device},
-            set(self.collection.attached(persistent=False)))
+            set(self.collection.attached()))
         self.assertEventFired(self.emitter, 'device-list-attached:testclass')
 
     def test_015_list_available(self):
@@ -147,6 +159,7 @@ class TC_01_DeviceManager(qubes.tests.QubesTestCase):
 
     def test_001_missing(self):
         device = TestDevice(self.emitter.app.domains['vm'], 'testdev')
-        self.manager['testclass'].attach(device)
+        assignment = qubes.devices.DeviceAssignment(backend_domain=device.backend_domain, ident=device.ident, persistent=True)
+        self.manager['testclass'].attach(assignment)
         self.assertEventFired(self.emitter, 'device-attach:testclass')
 

qubes/tests/tools/qvm_device.py

@@ -1,4 +1,4 @@
-# pylint: disable=protected-access,pointless-statement
+# pylint: disable=protected-access
 
 #
 # The Qubes OS Project, https://www.qubes-os.org/
@@ -21,6 +21,8 @@
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 #
 
+''' Tests for the `qvm-device` tool. '''
+
 import qubes
 import qubes.devices
 import qubes.tools.qvm_device
@@ -30,82 +32,86 @@ import qubes.tests.devices
 import qubes.tests.tools
 
 class TestNamespace(object):
+    ''' A mock object for `argparse.Namespace`.
+    '''  # pylint: disable=too-few-public-methods
+
     def __init__(self, app, domains=None, device=None):
         super(TestNamespace, self).__init__()
         self.app = app
         self.devclass = 'testclass'
+        self.persistent = True
         if domains:
             self.domains = domains
         if device:
             self.device = device
+            self.device_assignment = qubes.devices.DeviceAssignment(
+                backend_domain=self.device.backend_domain,
+                ident=self.device.ident, persistent=self.persistent)
 
 
 class TC_00_Actions(qubes.tests.QubesTestCase):
+    ''' Tests the output logic of the qvm-device tool '''
     def setUp(self):
         super(TC_00_Actions, self).setUp()
         self.app = qubes.tests.devices.TestApp()
+        def save():
+            ''' A mock method for simulating a successful save '''
+            return True
+        self.app.save = save
         self.vm1 = qubes.tests.devices.TestVM(self.app, 'vm1')
         self.vm2 = qubes.tests.devices.TestVM(self.app, 'vm2')
         self.device = self.vm2.device
 
     def test_000_list_all(self):
+        ''' List all exposed vm devices. No devices are attached to other
+            domains.
+        '''
         args = TestNamespace(self.app)
         with qubes.tests.tools.StdoutBuffer() as buf:
             qubes.tools.qvm_device.list_devices(args)
-            self.assertEventFired(self.vm1,
-                'device-list:testclass')
-            self.assertEventFired(self.vm2,
-                'device-list:testclass')
-            self.assertEventNotFired(self.vm1,
-                'device-list-attached:testclass')
-            self.assertEventNotFired(self.vm2,
-                'device-list-attached:testclass')
             self.assertEqual(
                 [x.rstrip() for x in buf.getvalue().splitlines()],
                 ['vm1:testdev  Description',
                  'vm2:testdev  Description']
             )
 
-    def test_001_list_one(self):
+    def test_001_list_persistent_attach(self):
+        ''' Attach the device exposed by the `vm2` to the `vm1` persistently.
+        '''
         args = TestNamespace(self.app, [self.vm1])
         # simulate attach
+        assignment = qubes.devices.DeviceAssignment(backend_domain=self.vm2,
+            ident=self.device.ident, persistent=True, frontend_domain=self.vm1)
+
         self.vm2.device.frontend_domain = self.vm1
-        self.vm1.devices['testclass']._set.add(self.device)
+        self.vm1.devices['testclass']._set.add(assignment)
         with qubes.tests.tools.StdoutBuffer() as buf:
             qubes.tools.qvm_device.list_devices(args)
-            self.assertEventFired(self.vm1,
-                'device-list-attached:testclass')
-            self.assertEventNotFired(self.vm1,
-                'device-list:testclass')
-            self.assertEventNotFired(self.vm2,
-                'device-list:testclass')
-            self.assertEventNotFired(self.vm2,
-                'device-list-attached:testclass')
             self.assertEqual(
                 buf.getvalue(),
-                'vm2:testdev  Description  vm1\n'
+                'vm1:testdev  Description\n'
+                'vm2:testdev  Description  vm1  vm1\n'
             )
 
-    def test_002_list_one_non_persistent(self):
+    def test_002_list_list_temp_attach(self):
+        ''' Attach the device exposed by the `vm2` to the `vm1`
+            non-persistently.
+        '''
         args = TestNamespace(self.app, [self.vm1])
         # simulate attach
+        assignment = qubes.devices.DeviceAssignment(backend_domain=self.vm2,
+            ident=self.device.ident, persistent=True, frontend_domain=self.vm1)
+
         self.vm2.device.frontend_domain = self.vm1
+        self.vm1.devices['testclass']._set.add(assignment)
         with qubes.tests.tools.StdoutBuffer() as buf:
             qubes.tools.qvm_device.list_devices(args)
-            self.assertEventFired(self.vm1,
-                'device-list-attached:testclass')
-            self.assertEventNotFired(self.vm1,
-                'device-list:testclass')
-            self.assertEventNotFired(self.vm2,
-                'device-list:testclass')
-            self.assertEventNotFired(self.vm2,
-                'device-list-attached:testclass')
-            self.assertEqual(
-                buf.getvalue(),
-                'vm2:testdev  Description  vm1\n'
-            )
+            self.assertEqual(buf.getvalue(),
+                            'vm1:testdev  Description\n'
+                            'vm2:testdev  Description  vm1  vm1\n')
 
     def test_010_attach(self):
+        ''' Test attach action '''
         args = TestNamespace(
             self.app,
             [self.vm1],
@@ -118,6 +124,7 @@ class TC_00_Actions(qubes.tests.QubesTestCase):
             'device-attach:testclass', kwargs={'device': self.device})
 
     def test_011_double_attach(self):
+        ''' Double attach should not be possible '''
         args = TestNamespace(
             self.app,
             [self.vm1],
@@ -128,6 +135,7 @@ class TC_00_Actions(qubes.tests.QubesTestCase):
             qubes.tools.qvm_device.attach_device(args)
 
     def test_020_detach(self):
+        ''' Test detach action '''
         args = TestNamespace(
             self.app,
             [self.vm1],
@@ -135,10 +143,12 @@ class TC_00_Actions(qubes.tests.QubesTestCase):
         )
         # simulate attach
         self.vm2.device.frontend_domain = self.vm1
-        self.vm1.devices['testclass']._set.add(self.device)
+        args.device_assignment.frontend_domain = self.vm1
+        self.vm1.devices['testclass']._set.add(args.device_assignment)
         qubes.tools.qvm_device.detach_device(args)
 
     def test_021_detach_not_attached(self):
+        ''' Invalid detach action should not be possible '''
         args = TestNamespace(
             self.app,
             [self.vm1],