adminvm.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org/
  3. #
  4. # Copyright (C) 2014-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  5. # Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  6. #
  7. # This library is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU Lesser General Public
  9. # License as published by the Free Software Foundation; either
  10. # version 2.1 of the License, or (at your option) any later version.
  11. #
  12. # This library is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15. # Lesser General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU Lesser General Public
  18. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  19. #
  20. import subprocess
  21. import unittest
  22. import unittest.mock
  23. import functools
  24. import asyncio
  25. import qubes
  26. import qubes.exc
  27. import qubes.vm
  28. import qubes.vm.adminvm
  29. import qubes.tests
  30. class TC_00_AdminVM(qubes.tests.QubesTestCase):
  31. def setUp(self):
  32. super().setUp()
  33. try:
  34. self.app = qubes.tests.vm.TestApp()
  35. with unittest.mock.patch.object(
  36. qubes.vm.adminvm.AdminVM, 'start_qdb_watch') as mock_qdb:
  37. self.vm = qubes.vm.adminvm.AdminVM(self.app,
  38. xml=None)
  39. mock_qdb.assert_called_once_with()
  40. self.addCleanup(self.cleanup_adminvm)
  41. except: # pylint: disable=bare-except
  42. if self.id().endswith('.test_000_init'):
  43. raise
  44. self.skipTest('setup failed')
  45. def tearDown(self) -> None:
  46. self.app.domains.clear()
  47. def add_vm(self, name, cls=qubes.vm.qubesvm.QubesVM, **kwargs):
  48. vm = cls(self.app, None,
  49. qid=kwargs.pop('qid', 1), name=qubes.tests.VMPREFIX + name,
  50. **kwargs)
  51. self.app.domains[vm.qid] = vm
  52. self.app.domains[vm.uuid] = vm
  53. self.app.domains[vm.name] = vm
  54. self.app.domains[vm] = vm
  55. self.addCleanup(vm.close)
  56. return vm
  57. @asyncio.coroutine
  58. def coroutine_mock(self, mock, *args, **kwargs):
  59. return mock(*args, **kwargs)
  60. def cleanup_adminvm(self):
  61. self.vm.close()
  62. del self.vm
  63. def test_000_init(self):
  64. pass
  65. def test_100_xid(self):
  66. self.assertEqual(self.vm.xid, 0)
  67. def test_101_libvirt_domain(self):
  68. with unittest.mock.patch.object(self.app, 'vmm') as mock_vmm:
  69. self.assertIsNotNone(self.vm.libvirt_domain)
  70. self.assertEqual(mock_vmm.mock_calls, [
  71. ('libvirt_conn.lookupByID', (0,), {}),
  72. ])
  73. def test_300_is_running(self):
  74. self.assertTrue(self.vm.is_running())
  75. def test_301_get_power_state(self):
  76. self.assertEqual(self.vm.get_power_state(), 'Running')
  77. def test_302_get_mem(self):
  78. self.assertGreater(self.vm.get_mem(), 0)
  79. @unittest.skip('mock object does not support this')
  80. def test_303_get_mem_static_max(self):
  81. self.assertGreater(self.vm.get_mem_static_max(), 0)
  82. def test_310_start(self):
  83. with self.assertRaises(qubes.exc.QubesException):
  84. self.vm.start()
  85. @unittest.skip('this functionality is undecided')
  86. def test_311_suspend(self):
  87. with self.assertRaises(qubes.exc.QubesException):
  88. self.vm.suspend()
  89. @unittest.mock.patch('asyncio.create_subprocess_exec')
  90. def test_700_run_service(self, mock_subprocess):
  91. func_mock = unittest.mock.Mock()
  92. mock_subprocess.side_effect = functools.partial(
  93. self.coroutine_mock, func_mock)
  94. self.add_vm('vm')
  95. with self.subTest('running'):
  96. self.loop.run_until_complete(self.vm.run_service('test.service'))
  97. func_mock.assert_called_once_with(
  98. '/usr/lib/qubes/qubes-rpc-multiplexer',
  99. 'test.service', 'dom0', 'name', 'dom0')
  100. func_mock.reset_mock()
  101. with self.subTest('other_user'):
  102. self.loop.run_until_complete(
  103. self.vm.run_service('test.service', user='other'))
  104. func_mock.assert_called_once_with(
  105. 'runuser', '-u', 'other', '--',
  106. '/usr/lib/qubes/qubes-rpc-multiplexer',
  107. 'test.service', 'dom0', 'name', 'dom0')
  108. func_mock.reset_mock()
  109. with self.subTest('other_source'):
  110. self.loop.run_until_complete(
  111. self.vm.run_service('test.service', source='test-inst-vm'))
  112. func_mock.assert_called_once_with(
  113. '/usr/lib/qubes/qubes-rpc-multiplexer',
  114. 'test.service', 'test-inst-vm', 'name', 'dom0')
  115. @unittest.mock.patch('qubes.vm.adminvm.AdminVM.run_service')
  116. def test_710_run_service_for_stdio(self, mock_run_service):
  117. func_mock = unittest.mock.Mock()
  118. mock_run_service.side_effect = functools.partial(
  119. self.coroutine_mock, func_mock)
  120. communicate_mock = unittest.mock.Mock()
  121. func_mock.return_value.communicate.side_effect = functools.partial(
  122. self.coroutine_mock, communicate_mock)
  123. communicate_mock.return_value = (b'stdout', b'stderr')
  124. func_mock.return_value.returncode = 0
  125. with self.subTest('default'):
  126. value = self.loop.run_until_complete(
  127. self.vm.run_service_for_stdio('test.service'))
  128. func_mock.assert_called_once_with(
  129. 'test.service',
  130. stdout=subprocess.PIPE,
  131. stderr=subprocess.PIPE,
  132. stdin=subprocess.PIPE)
  133. communicate_mock.assert_called_once_with(input=None)
  134. self.assertEqual(value, (b'stdout', b'stderr'))
  135. func_mock.reset_mock()
  136. communicate_mock.reset_mock()
  137. with self.subTest('with_input'):
  138. value = self.loop.run_until_complete(
  139. self.vm.run_service_for_stdio('test.service', input=b'abc'))
  140. func_mock.assert_called_once_with(
  141. 'test.service',
  142. stdout=subprocess.PIPE,
  143. stderr=subprocess.PIPE,
  144. stdin=subprocess.PIPE)
  145. communicate_mock.assert_called_once_with(input=b'abc')
  146. self.assertEqual(value, (b'stdout', b'stderr'))
  147. func_mock.reset_mock()
  148. communicate_mock.reset_mock()
  149. with self.subTest('error'):
  150. func_mock.return_value.returncode = 1
  151. with self.assertRaises(subprocess.CalledProcessError) as exc:
  152. self.loop.run_until_complete(
  153. self.vm.run_service_for_stdio('test.service'))
  154. func_mock.assert_called_once_with(
  155. 'test.service',
  156. stdout=subprocess.PIPE,
  157. stderr=subprocess.PIPE,
  158. stdin=subprocess.PIPE)
  159. communicate_mock.assert_called_once_with(input=None)
  160. self.assertEqual(exc.exception.returncode, 1)
  161. self.assertEqual(exc.exception.output, b'stdout')
  162. self.assertEqual(exc.exception.stderr, b'stderr')