devices.py 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
  1. # pylint: disable=protected-access,pointless-statement
  2. #
  3. # The Qubes OS Project, https://www.qubes-os.org/
  4. #
  5. # Copyright (C) 2015-2016 Joanna Rutkowska <joanna@invisiblethingslab.com>
  6. # Copyright (C) 2015-2016 Wojtek Porczyk <woju@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU General Public License as published by
  10. # the Free Software Foundation; either version 2 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU General Public License along
  19. # with this program; if not, write to the Free Software Foundation, Inc.,
  20. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21. #
  22. import qubes.devices
  23. import qubes.tests
  24. class TestDevice(qubes.devices.DeviceInfo):
  25. # pylint: disable=too-few-public-methods
  26. pass
  27. class TestVMCollection(dict):
  28. def __iter__(self):
  29. return iter(set(self.values()))
  30. class TestApp(object):
  31. # pylint: disable=too-few-public-methods
  32. def __init__(self):
  33. self.domains = TestVMCollection()
  34. class TestVM(qubes.tests.TestEmitter):
  35. def __init__(self, app, name, *args, **kwargs):
  36. super(TestVM, self).__init__(*args, **kwargs)
  37. self.app = app
  38. self.name = name
  39. self.device = TestDevice(self, 'testdev', 'Description')
  40. self.events_enabled = True
  41. self.devices = {
  42. 'testclass': qubes.devices.DeviceCollection(self, 'testclass')
  43. }
  44. self.app.domains[name] = self
  45. self.app.domains[self] = self
  46. self.running = False
  47. def __str__(self):
  48. return self.name
  49. @qubes.events.handler('device-list-attached:testclass')
  50. def dev_testclass_list_attached(self, event, persistent = False):
  51. for vm in self.app.domains:
  52. if vm.device.frontend_domain == self:
  53. yield (vm.device, {})
  54. @qubes.events.handler('device-list:testclass')
  55. def dev_testclass_list(self, event):
  56. yield self.device
  57. def is_halted(self):
  58. return not self.running
  59. class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
  60. def setUp(self):
  61. super().setUp()
  62. self.app = TestApp()
  63. self.emitter = TestVM(self.app, 'vm')
  64. self.app.domains['vm'] = self.emitter
  65. self.device = self.emitter.device
  66. self.collection = self.emitter.devices['testclass']
  67. self.assignment = qubes.devices.DeviceAssignment(
  68. backend_domain=self.device.backend_domain,
  69. ident=self.device.ident,
  70. persistent=True
  71. )
  72. def test_000_init(self):
  73. self.assertFalse(self.collection._set)
  74. def test_001_attach(self):
  75. self.loop.run_until_complete(self.collection.attach(self.assignment))
  76. self.assertEventFired(self.emitter, 'device-pre-attach:testclass')
  77. self.assertEventFired(self.emitter, 'device-attach:testclass')
  78. self.assertEventNotFired(self.emitter, 'device-pre-detach:testclass')
  79. self.assertEventNotFired(self.emitter, 'device-detach:testclass')
  80. def test_002_detach(self):
  81. self.loop.run_until_complete(self.collection.attach(self.assignment))
  82. self.loop.run_until_complete(self.collection.detach(self.assignment))
  83. self.assertEventFired(self.emitter, 'device-pre-attach:testclass')
  84. self.assertEventFired(self.emitter, 'device-attach:testclass')
  85. self.assertEventFired(self.emitter, 'device-pre-detach:testclass')
  86. self.assertEventFired(self.emitter, 'device-detach:testclass')
  87. def test_010_empty_detach(self):
  88. with self.assertRaises(LookupError):
  89. self.loop.run_until_complete(
  90. self.collection.detach(self.assignment))
  91. def test_011_double_attach(self):
  92. self.loop.run_until_complete(self.collection.attach(self.assignment))
  93. with self.assertRaises(qubes.devices.DeviceAlreadyAttached):
  94. self.loop.run_until_complete(
  95. self.collection.attach(self.assignment))
  96. def test_012_double_detach(self):
  97. self.loop.run_until_complete(self.collection.attach(self.assignment))
  98. self.loop.run_until_complete(self.collection.detach(self.assignment))
  99. with self.assertRaises(qubes.devices.DeviceNotAttached):
  100. self.loop.run_until_complete(
  101. self.collection.detach(self.assignment))
  102. def test_013_list_attached_persistent(self):
  103. self.assertEqual(set([]), set(self.collection.persistent()))
  104. self.loop.run_until_complete(self.collection.attach(self.assignment))
  105. self.assertEventFired(self.emitter, 'device-list-attached:testclass')
  106. self.assertEqual({self.device}, set(self.collection.persistent()))
  107. self.assertEqual({self.device},
  108. set(self.collection.persistent()))
  109. self.assertEqual(set([]),
  110. set(self.collection.attached()))
  111. def test_014_list_attached_non_persistent(self):
  112. self.assignment.persistent = False
  113. self.emitter.running = True
  114. self.loop.run_until_complete(self.collection.attach(self.assignment))
  115. # device-attach event not implemented, so manipulate object manually
  116. self.device.frontend_domain = self.emitter
  117. self.assertEqual({self.device},
  118. set(self.collection.attached()))
  119. self.assertEqual(set([]),
  120. set(self.collection.persistent()))
  121. self.assertEqual({self.device},
  122. set(self.collection.attached()))
  123. self.assertEventFired(self.emitter, 'device-list-attached:testclass')
  124. def test_015_list_available(self):
  125. self.assertEqual({self.device}, set(self.collection))
  126. self.assertEventFired(self.emitter, 'device-list:testclass')
  127. class TC_01_DeviceManager(qubes.tests.QubesTestCase):
  128. def setUp(self):
  129. super().setUp()
  130. self.app = TestApp()
  131. self.emitter = TestVM(self.app, 'vm')
  132. self.manager = qubes.devices.DeviceManager(self.emitter)
  133. def test_000_init(self):
  134. self.assertEqual(self.manager, {})
  135. def test_001_missing(self):
  136. device = TestDevice(self.emitter.app.domains['vm'], 'testdev')
  137. assignment = qubes.devices.DeviceAssignment(
  138. backend_domain=device.backend_domain,
  139. ident=device.ident,
  140. persistent=True)
  141. self.loop.run_until_complete(
  142. self.manager['testclass'].attach(assignment))
  143. self.assertEventFired(self.emitter, 'device-attach:testclass')