test_qube_manager.py 58 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465
  1. #!/usr/bin/python3
  2. #
  3. # The Qubes OS Project, https://www.qubes-os.org/
  4. #
  5. # Copyright (C) 2016 Marta Marczykowska-Górecka
  6. # <marmarta@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 asyncio
  23. import contextlib
  24. import logging.handlers
  25. import unittest
  26. import unittest.mock
  27. import subprocess
  28. import datetime
  29. import time
  30. from PyQt5 import QtTest, QtCore, QtWidgets
  31. from PyQt5.QtCore import (Qt, QSize)
  32. from PyQt5.QtGui import (QIcon)
  33. from qubesadmin import Qubes, events, exc
  34. import qubesmanager.qube_manager as qube_manager
  35. from qubesmanager.tests import init_qtapp
  36. icon_size = QSize(24, 24)
  37. class QubeManagerTest(unittest.TestCase):
  38. def setUp(self):
  39. super(QubeManagerTest, self).setUp()
  40. self.qtapp, self.loop = init_qtapp()
  41. self.mock_qprogress = unittest.mock.patch(
  42. 'PyQt5.QtWidgets.QProgressDialog')
  43. self.mock_qprogress.start()
  44. self.addCleanup(self.mock_qprogress.stop)
  45. self.qapp = Qubes()
  46. self.dispatcher = events.EventsDispatcher(self.qapp)
  47. self.dialog = qube_manager.VmManagerWindow(
  48. self.qtapp, self.qapp, self.dispatcher)
  49. def test_000_window_loads(self):
  50. self.assertTrue(self.dialog.table is not None, "Window did not load")
  51. def test_001_correct_vms_listed(self):
  52. vms_in_table = []
  53. for row in range(self.dialog.table.model().rowCount()):
  54. vm = self._get_table_vm(row)
  55. self.assertIsNotNone(vm)
  56. vms_in_table.append(vm.name)
  57. # check that name is listed correctly
  58. name_item = self._get_table_item(row, "Name")
  59. self.assertEqual(name_item, vm.name,
  60. "Incorrect VM name for {}".format(vm.name))
  61. actual_vms = [vm.name for vm in self.qapp.domains]
  62. self.assertEqual(len(vms_in_table), len(actual_vms),
  63. "Incorrect number of VMs loaded")
  64. self.assertListEqual(sorted(vms_in_table), sorted(actual_vms),
  65. "Incorrect VMs loaded")
  66. def test_002_correct_template_listed(self):
  67. for row in range(self.dialog.table.model().rowCount()):
  68. vm = self._get_table_vm(row)
  69. # check that template is listed correctly
  70. template_item = self._get_table_item(row, "Template")
  71. if getattr(vm, "template", None):
  72. self.assertEqual(vm.template,
  73. template_item,
  74. "Incorrect template for {}".format(vm.name))
  75. else:
  76. self.assertEqual(vm.klass, template_item,
  77. "Incorrect class for {}".format(vm.name))
  78. def test_003_correct_netvm_listed(self):
  79. for row in range(self.dialog.table.model().rowCount()):
  80. vm = self._get_table_vm(row)
  81. # check that netvm is listed correctly
  82. netvm_item = self._get_table_item(row, "NetVM")
  83. netvm_value = getattr(vm, "netvm", None)
  84. if not netvm_value:
  85. netvm_value = "n/a"
  86. if netvm_value and hasattr(vm, "netvm") \
  87. and vm.property_is_default("netvm"):
  88. netvm_value = "default ({})".format(netvm_value)
  89. self.assertEqual(netvm_value,
  90. netvm_item,
  91. "Incorrect netvm for {}".format(vm.name))
  92. def test_004_correct_disk_usage_listed(self):
  93. for row in range(self.dialog.table.model().rowCount()):
  94. vm = self._get_table_vm(row)
  95. size_item = self._get_table_item(row, "Disk Usage")
  96. if vm.klass == 'AdminVM':
  97. size_value = "n/a"
  98. else:
  99. size_value = round(vm.get_disk_utilization() / (1024 * 1024), 2)
  100. size_value = str(size_value) + " MiB"
  101. self.assertEqual(size_value,
  102. size_item,
  103. "Incorrect size for {}".format(vm.name))
  104. def test_005_correct_internal_listed(self):
  105. for row in range(self.dialog.table.model().rowCount()):
  106. vm = self._get_table_vm(row)
  107. internal_item = self._get_table_item(row, "Internal")
  108. internal_value = "Yes" if vm.features.get('internal', False) else ""
  109. self.assertEqual(internal_item, internal_value,
  110. "Incorrect internal value for {}".format(vm.name))
  111. def test_006_correct_ip_listed(self):
  112. for row in range(self.dialog.table.model().rowCount()):
  113. vm = self._get_table_vm(row)
  114. ip_item = self._get_table_item(row, "IP")
  115. if hasattr(vm, 'ip'):
  116. ip_value = getattr(vm, 'ip')
  117. ip_value = "" if ip_value is None else ip_value
  118. else:
  119. ip_value = "n/a"
  120. self.assertEqual(ip_value, ip_item,
  121. "Incorrect ip value for {}".format(vm.name))
  122. def test_007_incl_in_backups_listed(self):
  123. for row in range(self.dialog.table.model().rowCount()):
  124. vm = self._get_table_vm(row)
  125. incl_backups_item = self._get_table_item(row, "Include in backups")
  126. incl_backups_value = getattr(vm, 'include_in_backups', False)
  127. incl_backups_value = "Yes" if incl_backups_value else ""
  128. self.assertEqual(
  129. incl_backups_value, incl_backups_item,
  130. "Incorrect include in backups value for {}".format(vm.name))
  131. def test_008_last_backup_listed(self):
  132. for row in range(self.dialog.table.model().rowCount()):
  133. vm = self._get_table_vm(row)
  134. last_backup_item = self._get_table_item(row, "Last backup")
  135. last_backup_value = getattr(vm, 'backup_timestamp', None)
  136. if last_backup_value:
  137. last_backup_value = str(
  138. datetime.datetime.fromtimestamp(last_backup_value))
  139. self.assertEqual(
  140. last_backup_value, last_backup_item,
  141. "Incorrect last backup value for {}".format(vm.name))
  142. def test_009_def_dispvm_listed(self):
  143. for row in range(self.dialog.table.model().rowCount()):
  144. vm = self._get_table_vm(row)
  145. def_dispvm_item = self._get_table_item(row, "Default DispVM")
  146. if vm.property_is_default("default_dispvm"):
  147. def_dispvm_value = "default ({})".format(
  148. self.qapp.default_dispvm)
  149. else:
  150. def_dispvm_value = getattr(vm, "default_dispvm", None)
  151. self.assertEqual(
  152. def_dispvm_value, def_dispvm_item,
  153. "Incorrect default dispvm value for {}".format(vm.name))
  154. def test_010_is_dvm_template_listed(self):
  155. for row in range(self.dialog.table.model().rowCount()):
  156. vm = self._get_table_vm(row)
  157. is_dvm_template_item = self._get_table_item(row, "Is DVM Template")
  158. is_dvm_template_value = "Yes" if \
  159. getattr(vm, "template_for_dispvms", False) else ""
  160. self.assertEqual(
  161. is_dvm_template_value, is_dvm_template_item,
  162. "Incorrect is DVM template value for {}".format(vm.name))
  163. def test_011_is_label_correct(self):
  164. for row in range(self.dialog.table.model().rowCount()):
  165. vm = self._get_table_vm(row)
  166. icon = QIcon.fromTheme(vm.label.icon)
  167. icon = icon.pixmap(icon_size)
  168. label_pixmap = self._get_table_item(row, "Label", Qt.DecorationRole)
  169. self.assertEqual(label_pixmap.toImage(), icon.toImage())
  170. def test_012_is_state_correct(self):
  171. for row in range(self.dialog.table.model().rowCount()):
  172. vm = self._get_table_vm(row)
  173. displayed_power_state = self._get_table_item(row, "State")['power']
  174. self.assertEqual(
  175. displayed_power_state, vm.get_power_state(),
  176. "Wrong power state displayed for {}".format(vm.name))
  177. def test_013_incorrect_settings_file(self):
  178. mock_settings = unittest.mock.MagicMock(spec=QtCore.QSettings)
  179. settings_result_dict = {"view/sort_column": "Cthulhu",
  180. "view/sort_order": "Fhtagn",
  181. "view/menubar_visible": "R'lyeh"
  182. }
  183. mock_settings.side_effect = (
  184. lambda x, *args, **kwargs: settings_result_dict.get(x))
  185. with unittest.mock.patch('PyQt5.QtCore.QSettings.value',
  186. mock_settings),\
  187. unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')\
  188. as mock_warning:
  189. self.dialog = qube_manager.VmManagerWindow(
  190. self.qtapp, self.qapp, self.dispatcher)
  191. self.assertEqual(mock_warning.call_count, 1)
  192. def test_100_sorting(self):
  193. col = self.dialog.qubes_model.columns_indices.index("Template")
  194. self.dialog.table.sortByColumn(col, QtCore.Qt.AscendingOrder)
  195. self.__check_sorting("Template")
  196. col = self.dialog.qubes_model.columns_indices.index("Name")
  197. self.dialog.table.sortByColumn(col, QtCore.Qt.AscendingOrder)
  198. self.__check_sorting("Name")
  199. @unittest.mock.patch('qubesmanager.qube_manager.QSettings.setValue')
  200. def test_101_hide_column(self, mock_settings):
  201. model = self.dialog.qubes_model
  202. action_no = model.columns_indices.index('Is DVM Template')
  203. self.dialog.menu_view.actions()[action_no].trigger()
  204. mock_settings.assert_called_with('columns/Is DVM Template', True)
  205. self.dialog.menu_view.actions()[action_no].trigger()
  206. mock_settings.assert_called_with('columns/Is DVM Template', False)
  207. @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
  208. def test_200_vm_open_settings(self, mock_window):
  209. selected_vm = self._select_non_admin_vm()
  210. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  211. widget = self.dialog.toolbar.widgetForAction(
  212. self.dialog.action_settings)
  213. QtTest.QTest.mouseClick(widget,
  214. QtCore.Qt.LeftButton)
  215. mock_window.assert_called_once_with(
  216. selected_vm, qapp=self.qtapp, init_page="basic", qubesapp=self.qapp)
  217. def test_201_vm_open_settings_admin(self):
  218. self._select_admin_vm()
  219. self.assertFalse(self.dialog.action_settings.isEnabled(),
  220. "Settings not disabled for admin VM")
  221. self.assertFalse(self.dialog.action_editfwrules.isEnabled(),
  222. "Editfw not disabled for admin VM")
  223. self.assertFalse(self.dialog.action_appmenus.isEnabled(),
  224. "Appmenus not disabled for admin VM")
  225. @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
  226. def test_202_vm_open_firewall(self, mock_window):
  227. selected_vm = self._select_non_admin_vm()
  228. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  229. widget = self.dialog.toolbar.widgetForAction(
  230. self.dialog.action_editfwrules)
  231. QtTest.QTest.mouseClick(widget,
  232. QtCore.Qt.LeftButton)
  233. mock_window.assert_called_once_with(
  234. selected_vm, qapp=self.qtapp, init_page="firewall",
  235. qubesapp=self.qapp)
  236. @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
  237. def test_203_vm_open_apps(self, mock_window):
  238. selected_vm = self._select_non_admin_vm()
  239. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  240. widget = self.dialog.toolbar.widgetForAction(
  241. self.dialog.action_appmenus)
  242. QtTest.QTest.mouseClick(widget,
  243. QtCore.Qt.LeftButton)
  244. mock_window.assert_called_once_with(
  245. selected_vm, qapp=self.qtapp, init_page="applications",
  246. qubesapp=self.qapp)
  247. def test_204_vm_keyboard(self):
  248. selected_vm = self._select_non_admin_vm(running=True)
  249. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  250. widget = self.dialog.toolbar.widgetForAction(
  251. self.dialog.action_set_keyboard_layout)
  252. with unittest.mock.patch.object(selected_vm, 'run') as mock_run:
  253. QtTest.QTest.mouseClick(widget,
  254. QtCore.Qt.LeftButton)
  255. mock_run.assert_called_once_with("qubes-change-keyboard-layout")
  256. def test_205_vm_keyboard_not_running(self):
  257. selected_vm = self._select_non_admin_vm(running=False)
  258. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  259. widget = self.dialog.toolbar.widgetForAction(
  260. self.dialog.action_set_keyboard_layout)
  261. with unittest.mock.patch.object(selected_vm, 'run') as mock_run:
  262. QtTest.QTest.mouseClick(widget,
  263. QtCore.Qt.LeftButton)
  264. self.assertEqual(mock_run.call_count, 0,
  265. "Keyboard change called on a halted VM")
  266. def test_206_dom0_keyboard(self):
  267. self._select_admin_vm()
  268. self.assertFalse(self.dialog.action_set_keyboard_layout.isEnabled())
  269. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  270. return_value=QtWidgets.QMessageBox.Yes)
  271. def test_207_update_vm_not_running(self, _):
  272. selected_vm = self._select_templatevm(running=False)
  273. self.assertIsNotNone(selected_vm, "No valid template VM found")
  274. widget = self.dialog.toolbar.widgetForAction(
  275. self.dialog.action_updatevm)
  276. with unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread') \
  277. as mock_update:
  278. QtTest.QTest.mouseClick(widget,
  279. QtCore.Qt.LeftButton)
  280. mock_update.assert_called_once_with(selected_vm)
  281. mock_update().start.assert_called_once_with()
  282. def test_208_update_vm_admin(self):
  283. selected_vm = self._select_admin_vm()
  284. self.assertIsNotNone(selected_vm, "No valid admin VM found")
  285. widget = self.dialog.toolbar.widgetForAction(
  286. self.dialog.action_updatevm)
  287. with unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread') \
  288. as mock_update:
  289. QtTest.QTest.mouseClick(widget,
  290. QtCore.Qt.LeftButton)
  291. mock_update.assert_called_once_with(selected_vm)
  292. mock_update().start.assert_called_once_with()
  293. @unittest.mock.patch("PyQt5.QtWidgets.QInputDialog.getText",
  294. return_value=("command to run", True))
  295. def test_209_run_command_in_vm(self, _):
  296. selected_vm = self._select_non_admin_vm()
  297. self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
  298. with unittest.mock.patch('qubesmanager.qube_manager.RunCommandThread') \
  299. as mock_thread:
  300. self.dialog.action_run_command_in_vm.trigger()
  301. mock_thread.assert_called_once_with(selected_vm, "command to run")
  302. mock_thread().finished.connect.assert_called_once_with(
  303. self.dialog.clear_threads)
  304. mock_thread().start.assert_called_once_with()
  305. def test_210_run_command_in_adminvm(self):
  306. self._select_admin_vm()
  307. self.assertFalse(self.dialog.action_run_command_in_vm.isEnabled(),
  308. "Should not be able to run commands for dom0")
  309. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.warning")
  310. def test_211_pausevm(self, mock_warn):
  311. selected_vm = self._select_non_admin_vm(running=True)
  312. self.assertTrue(self.dialog.action_pausevm.isEnabled(),
  313. "Pause not enabled for a running VM")
  314. with unittest.mock.patch.object(selected_vm, 'pause') as mock_pause:
  315. self.dialog.action_pausevm.trigger()
  316. mock_pause.assert_called_once_with()
  317. mock_pause.side_effect = exc.QubesException('Error')
  318. self.dialog.action_pausevm.trigger()
  319. self.assertEqual(mock_warn.call_count, 1)
  320. def test_212_resumevm(self):
  321. selected_vm = self._select_non_admin_vm(running=False)
  322. with unittest.mock.patch.object(selected_vm, 'get_power_state')\
  323. as mock_state, \
  324. unittest.mock.patch.object(selected_vm, 'unpause')\
  325. as mock_unpause:
  326. mock_state.return_value = 'Paused'
  327. self.dialog.action_resumevm.trigger()
  328. mock_unpause.assert_called_once_with()
  329. with unittest.mock.patch('qubesmanager.qube_manager.StartVMThread') \
  330. as mock_thread:
  331. self.dialog.action_resumevm.trigger()
  332. mock_thread.assert_called_once_with(selected_vm)
  333. mock_thread().finished.connect.assert_called_once_with(
  334. self.dialog.clear_threads)
  335. mock_thread().start.assert_called_once_with()
  336. def test_213_resume_running_vm(self):
  337. self._select_non_admin_vm(running=True)
  338. self.assertFalse(self.dialog.action_resumevm.isEnabled())
  339. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  340. return_value=QtWidgets.QMessageBox.Yes)
  341. @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
  342. @unittest.mock.patch('qubesmanager.qube_manager.VmShutdownMonitor')
  343. def test_214_shutdownvm(self, mock_monitor, mock_timer, _):
  344. selected_vm = self._select_non_admin_vm(running=True)
  345. with unittest.mock.patch.object(selected_vm, 'shutdown')\
  346. as mock_shutdown:
  347. self.dialog.action_shutdownvm.trigger()
  348. mock_shutdown.assert_called_once_with()
  349. mock_monitor.assert_called_once_with(
  350. selected_vm,
  351. unittest.mock.ANY, unittest.mock.ANY,
  352. unittest.mock.ANY, unittest.mock.ANY)
  353. mock_timer.assert_called_once_with(unittest.mock.ANY,
  354. unittest.mock.ANY)
  355. def test_215_shutdown_halted_vm(self):
  356. self._select_non_admin_vm(running=False)
  357. self.assertFalse(self.dialog.action_shutdownvm.isEnabled())
  358. @unittest.mock.patch('qubesmanager.create_new_vm.NewVmDlg')
  359. def test_216_create_vm(self, mock_new_vm):
  360. action = self.dialog.action_createvm
  361. self.assertTrue(action.isEnabled())
  362. action.trigger()
  363. self.assertEqual(mock_new_vm.call_count, 1,
  364. "Create New VM window did not appear")
  365. def test_217_remove_admin_vm(self):
  366. self._select_admin_vm()
  367. self.assertFalse(self.dialog.action_removevm.isEnabled())
  368. @unittest.mock.patch("qubesmanager.qube_manager.QMessageBox")
  369. @unittest.mock.patch('qubesadmin.utils.vm_dependencies')
  370. def test_218_remove_vm_dependencies(self, mock_dependencies, mock_msgbox):
  371. mock_vm = unittest.mock.Mock(spec=['name'],
  372. **{'name.return_value': 'test-vm'})
  373. mock_dependencies.return_value = [(mock_vm, "test_prop")]
  374. action = self.dialog.action_removevm
  375. self._select_non_admin_vm()
  376. action.trigger()
  377. mock_msgbox().show.assert_called_with()
  378. @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
  379. @unittest.mock.patch("PyQt5.QtWidgets.QInputDialog.getText")
  380. @unittest.mock.patch('qubesadmin.utils.vm_dependencies')
  381. def test_219_remove_vm_no_depencies(
  382. self, mock_dependencies, mock_input, mock_warning):
  383. action = self.dialog.action_removevm
  384. selected_vm = self._select_non_admin_vm(running=False)
  385. # test with no dependencies
  386. mock_dependencies.return_value = None
  387. with unittest.mock.patch('qubesmanager.common_threads.RemoveVMThread')\
  388. as mock_thread:
  389. mock_input.return_value = (selected_vm, False)
  390. action.trigger()
  391. self.assertEqual(mock_thread.call_count, 0,
  392. "VM removed despite user clicking 'cancel")
  393. mock_input.return_value = ("wrong_name", True)
  394. action.trigger()
  395. self.assertEqual(mock_warning.call_count, 1)
  396. self.assertEqual(mock_thread.call_count, 0,
  397. "VM removed despite user not confirming the name")
  398. mock_input.return_value = (selected_vm.name, True)
  399. action.trigger()
  400. mock_thread.assert_called_once_with(selected_vm)
  401. mock_thread().finished.connect.assert_called_once_with(
  402. self.dialog.clear_threads)
  403. mock_thread().start.assert_called_once_with()
  404. def test_220_restartvm_halted_vm(self):
  405. self._select_non_admin_vm(running=False)
  406. self.assertFalse(self.dialog.action_restartvm.isEnabled())
  407. @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
  408. @unittest.mock.patch('qubesmanager.qube_manager.VmShutdownMonitor')
  409. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  410. return_value=QtWidgets.QMessageBox.Yes)
  411. def test_221_restartvm_running_vm(self, _msgbox, mock_monitor, _qtimer):
  412. selected_vm = self._select_non_admin_vm(running=True)
  413. action = self.dialog.action_restartvm
  414. # currently the VM is running
  415. with unittest.mock.patch.object(selected_vm, 'shutdown')\
  416. as mock_shutdown:
  417. action.trigger()
  418. mock_shutdown.assert_called_once_with()
  419. mock_monitor.assert_called_once_with(
  420. selected_vm, unittest.mock.ANY,
  421. unittest.mock.ANY, True, unittest.mock.ANY)
  422. @unittest.mock.patch('qubesmanager.qube_manager.StartVMThread')
  423. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  424. return_value=QtWidgets.QMessageBox.Yes)
  425. def test_222_restartvm_shutdown_meantime(self, _, mock_thread):
  426. selected_vm = self._select_non_admin_vm(running=True)
  427. action = self.dialog.action_restartvm
  428. # it was shutdown in the meantime
  429. with unittest.mock.patch.object(
  430. selected_vm, 'is_running', **{'return_value': False}):
  431. action.trigger()
  432. mock_thread.assert_called_once_with(selected_vm)
  433. mock_thread().finished.connect.assert_called_once_with(
  434. self.dialog.clear_threads)
  435. mock_thread().start.assert_called_once_with()
  436. @unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread')
  437. def test_223_updatevm_running(self, mock_thread):
  438. selected_vm = self._select_non_admin_vm(running=True)
  439. self.dialog.action_updatevm.trigger()
  440. mock_thread.assert_called_once_with(selected_vm)
  441. mock_thread().finished.connect.assert_called_once_with(
  442. self.dialog.clear_threads)
  443. mock_thread().start.assert_called_once_with()
  444. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  445. return_value=QtWidgets.QMessageBox.Yes)
  446. @unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread')
  447. def test_224_updatevm_halted(self, mock_thread, _):
  448. selected_vm = self._select_non_admin_vm(running=False)
  449. self.dialog.action_updatevm.trigger()
  450. mock_thread.assert_called_once_with(selected_vm)
  451. mock_thread().finished.connect.assert_called_once_with(
  452. self.dialog.clear_threads)
  453. mock_thread().start.assert_called_once_with()
  454. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  455. return_value=QtWidgets.QMessageBox.Yes)
  456. def test_224_killvm(self, _):
  457. selected_vm = self._select_non_admin_vm(running=True)
  458. action = self.dialog.action_killvm
  459. with unittest.mock.patch.object(selected_vm, 'kill') as mock_kill:
  460. action.trigger()
  461. mock_kill.assert_called_once_with()
  462. @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
  463. return_value=QtWidgets.QMessageBox.Cancel)
  464. def test_225_killvm_cancel(self, _):
  465. selected_vm = self._select_non_admin_vm(running=True)
  466. action = self.dialog.action_killvm
  467. with unittest.mock.patch.object(selected_vm, 'kill') as mock_kill:
  468. action.trigger()
  469. self.assertEqual(mock_kill.call_count, 0,
  470. "Ignored Cancel on kill VM")
  471. @unittest.mock.patch('qubesmanager.global_settings.GlobalSettingsWindow')
  472. def test_226_global_settings(self, mock_settings):
  473. self._select_non_admin_vm()
  474. self.dialog.action_global_settings.trigger()
  475. self.assertEqual(mock_settings.call_count, 1,
  476. "Global Settings not opened")
  477. self._select_admin_vm()
  478. self.dialog.action_global_settings.trigger()
  479. self.assertEqual(mock_settings.call_count, 2,
  480. "Global Settings not opened for the second time")
  481. @unittest.mock.patch('qubesmanager.backup.BackupVMsWindow')
  482. def test_227_backup(self, mock_backup):
  483. self.dialog.action_backup.trigger()
  484. self.assertTrue(self.dialog.action_backup.isEnabled())
  485. self.assertEqual(mock_backup.call_count, 1,
  486. "Backup window does not appear")
  487. @unittest.mock.patch('qubesmanager.restore.RestoreVMsWindow')
  488. def test_228_restore(self, mock_restore):
  489. self.dialog.action_restore.trigger()
  490. self.assertTrue(self.dialog.action_restore.isEnabled())
  491. self.assertEqual(mock_restore.call_count, 1,
  492. "Backup window does not appear")
  493. @unittest.mock.patch('qubesmanager.qube_manager.AboutDialog')
  494. def test_229_about_qubes(self, mock_about):
  495. self.assertTrue(self.dialog.action_about_qubes.isEnabled())
  496. self.dialog.action_about_qubes.trigger()
  497. self.assertEqual(
  498. mock_about.call_count, 1, "About window does not appear")
  499. def test_230_exit_action(self):
  500. self.assertTrue(self.dialog.action_exit.isEnabled())
  501. with unittest.mock.patch.object(self.dialog, 'close') as mock_close:
  502. self.dialog.action_exit.trigger()
  503. mock_close.assert_called_once_with()
  504. @unittest.mock.patch('subprocess.check_call')
  505. def test_231_template_manager(self, mock_subprocess):
  506. self.assertTrue(self.dialog.action_manage_templates.isEnabled())
  507. self.dialog.action_manage_templates.trigger()
  508. mock_subprocess.assert_called_once_with('qubes-template-manager')
  509. @unittest.mock.patch('qubesmanager.clone_vm.CloneVMDlg')
  510. def test_232_clonevm(self, mock_clone):
  511. action = self.dialog.action_clonevm
  512. self._select_admin_vm()
  513. self.assertFalse(action.isEnabled())
  514. selected_vm = self._select_non_admin_vm()
  515. self.assertTrue(action.isEnabled())
  516. action.trigger()
  517. mock_clone.assert_called_once_with(self.qtapp, self.qapp,
  518. src_vm=selected_vm)
  519. def test_233_search_action(self):
  520. self.qtapp.setActiveWindow(self.dialog.searchbox)
  521. self.dialog.action_search.trigger()
  522. self.assertTrue(self.dialog.searchbox.hasFocus())
  523. # input text
  524. self.dialog.searchbox.setText("sys")
  525. # click outside the widget
  526. QtTest.QTest.mouseClick(self.dialog.table, QtCore.Qt.LeftButton)
  527. # click the widget, check if it is correctly activated and the whole
  528. # text was selected
  529. QtTest.QTest.mouseClick(self.dialog.searchbox, QtCore.Qt.LeftButton)
  530. self.assertTrue(self.dialog.searchbox.hasFocus())
  531. self.assertEqual(self.dialog.searchbox.selectedText(), "sys")
  532. def test_234_searchbox(self):
  533. # look for sys
  534. self.dialog.searchbox.setText("sys")
  535. expected_number = \
  536. len([vm for vm in self.qapp.domains if "sys" in vm.name])
  537. actual_number = self._count_visible_table_rows()
  538. self.assertEqual(expected_number, actual_number,
  539. "Incorrect number of vms shown for 'sys'")
  540. # clear search
  541. self.dialog.searchbox.setText("")
  542. expected_number = len([vm for vm in self.qapp.domains])
  543. actual_number = self._count_visible_table_rows()
  544. self.assertEqual(expected_number, actual_number,
  545. "Incorrect number of vms shown for cleared search box")
  546. def test_235_hide_show_toolbars(self):
  547. with unittest.mock.patch('PyQt5.QtCore.QSettings.setValue')\
  548. as mock_setvalue:
  549. self.dialog.action_menubar.trigger()
  550. mock_setvalue.assert_called_with('view/menubar_visible', False)
  551. self.dialog.action_toolbar.trigger()
  552. mock_setvalue.assert_called_with('view/toolbar_visible', False)
  553. self.assertFalse(self.dialog.menubar.isVisible(),
  554. "Menubar not hidden correctly")
  555. self.assertFalse(self.dialog.toolbar.isVisible(),
  556. "Toolbar not hidden correctly")
  557. def test_236_clear_searchbox(self):
  558. self.dialog.searchbox.setText("text")
  559. self.assertEqual(self.dialog.searchbox.text(), "text")
  560. QtTest.QTest.keyPress(self.dialog, QtCore.Qt.Key_Escape)
  561. self.assertEqual(self.dialog.searchbox.text(), "",
  562. "Escape failed to clear searchbox")
  563. expected_number = len([vm for vm in self.qapp.domains])
  564. actual_number = self._count_visible_table_rows()
  565. self.assertEqual(expected_number, actual_number,
  566. "Incorrect number of vms shown for cleared search box")
  567. @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.information')
  568. @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
  569. def test_300_clear_threads(self, mock_warning, mock_info):
  570. mock_thread_finished_ok = unittest.mock.Mock(
  571. spec=['isFinished', 'msg', 'msg_is_success'],
  572. msg=None, msg_is_success=False,
  573. **{'isFinished.return_value': True})
  574. mock_thread_not_finished = unittest.mock.Mock(
  575. spec=['isFinished', 'msg', 'msg_is_success'],
  576. msg=None, msg_is_success=False,
  577. **{'isFinished.return_value': False})
  578. mock_thread_finished_error = unittest.mock.Mock(
  579. spec=['isFinished', 'msg', 'msg_is_success'],
  580. msg=("Error", "Error"), msg_is_success=False,
  581. **{'isFinished.return_value': True})
  582. mock_thread_fin_error_success = unittest.mock.Mock(
  583. spec=['isFinished', 'msg', 'msg_is_success'],
  584. msg=("Done", "Done"), msg_is_success=True,
  585. **{'isFinished.return_value': True})
  586. # single finished thread
  587. self.dialog.threads_list = [mock_thread_not_finished,
  588. mock_thread_finished_ok]
  589. self.dialog.clear_threads()
  590. self.assertEqual(mock_warning.call_count, 0)
  591. self.assertEqual(mock_info.call_count, 0)
  592. self.assertEqual(len(self.dialog.threads_list), 1)
  593. # an error thread and some in-progress ones
  594. self.dialog.threads_list = [mock_thread_not_finished,
  595. mock_thread_not_finished,
  596. mock_thread_finished_error]
  597. self.dialog.clear_threads()
  598. self.assertEqual(mock_warning.call_count, 1)
  599. self.assertEqual(mock_info.call_count, 0)
  600. self.assertEqual(len(self.dialog.threads_list), 2)
  601. # an error-success thread and some in-progress ones
  602. self.dialog.threads_list = [mock_thread_not_finished,
  603. mock_thread_not_finished,
  604. mock_thread_fin_error_success,
  605. mock_thread_finished_error]
  606. self.dialog.clear_threads()
  607. self.assertEqual(mock_warning.call_count, 1)
  608. self.assertEqual(mock_info.call_count, 1)
  609. self.assertEqual(len(self.dialog.threads_list), 3)
  610. def test_400_event_domain_added(self):
  611. number_of_vms = self.dialog.table.model().rowCount()
  612. self.addCleanup(subprocess.call, ["qvm-remove", "-f", "test-vm"])
  613. self._run_command_and_process_events(
  614. ["qvm-create", "--label", "red", "test-vm"])
  615. # a single row was added to the table
  616. self.assertEqual(self.dialog.table.model().rowCount(), number_of_vms+1)
  617. # table contains the correct vms
  618. vms_in_table = self._create_set_of_current_vms()
  619. vms_in_system = set([vm.name for vm in self.qapp.domains])
  620. self.assertEqual(vms_in_table, vms_in_system, "Table not updated "
  621. "correctly after add")
  622. # check if sorting works
  623. self.__check_sorting("Name")
  624. # try opening settings for the added vm
  625. for row in range(self.dialog.table.model().rowCount()):
  626. name = self._get_table_item(row, "Name")
  627. if name == "test-vm":
  628. index = self.dialog.table.model().index(row, 0)
  629. self.dialog.table.setCurrentIndex(index)
  630. break
  631. with unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')\
  632. as mock_settings:
  633. self.dialog.action_settings.trigger()
  634. mock_settings.assert_called_once_with(
  635. self.qapp.domains["test-vm"], qapp = self.qtapp,
  636. init_page = "basic", qubesapp = self.qapp)
  637. def test_401_event_domain_removed(self):
  638. initial_vms = self._create_set_of_current_vms()
  639. self._run_command_and_process_events(
  640. ["qvm-create", "--label", "red", "test-vm"])
  641. current_vms = self._create_set_of_current_vms()
  642. self.assertEqual(len(initial_vms) + 1, len(current_vms))
  643. self._run_command_and_process_events(
  644. ["qvm-remove", "--force", "test-vm"])
  645. current_vms = self._create_set_of_current_vms()
  646. self.assertEqual(initial_vms, current_vms)
  647. # check if sorting works
  648. self.__check_sorting("Name")
  649. def test_403_event_dispvm_added(self):
  650. initial_vms = self._create_set_of_current_vms()
  651. dispvm_template = None
  652. for vm in self.qapp.domains:
  653. if getattr(vm, "template_for_dispvms", False):
  654. dispvm_template = vm.name
  655. break
  656. self.assertIsNotNone(dispvm_template,
  657. "Cannot find a template for dispVMs")
  658. # this requires very long timeout, because it takes time for the
  659. # dispvm to vanish
  660. self._run_command_and_process_events(
  661. ["qvm-run", "--dispvm", dispvm_template, "true"], timeout=60)
  662. final_vms = self._create_set_of_current_vms()
  663. self.assertEqual(initial_vms, final_vms,
  664. "Failed handling of a created-and-removed dispvm")
  665. def test_404_crashing_dispvm(self):
  666. initial_vms = self._create_set_of_current_vms()
  667. dispvm_template = None
  668. for vm in self.qapp.domains:
  669. if getattr(vm, "template_for_dispvms", False):
  670. dispvm_template = vm.name
  671. break
  672. self.assertIsNotNone(dispvm_template,
  673. "Cannot find a template for dispVMs")
  674. current_memory = getattr(self.qapp.domains[dispvm_template], "memory")
  675. self.addCleanup(
  676. subprocess.call,
  677. ["qvm-prefs", dispvm_template, "memory", str(current_memory)])
  678. subprocess.check_call(
  679. ["qvm-prefs", dispvm_template, "memory", "600000"])
  680. self._run_command_and_process_events(
  681. ["qvm-run", "--dispvm", dispvm_template, "true"], timeout=30)
  682. final_vms = self._create_set_of_current_vms()
  683. self.assertEqual(initial_vms, final_vms,
  684. "Failed handling of dispvm that crashed on start")
  685. def test_405_prop_change_label(self):
  686. target_vm_name = "work"
  687. vm_row = self._find_vm_row(target_vm_name)
  688. current_label = self._get_table_item(vm_row, "Label", Qt.DecorationRole)
  689. self.addCleanup(
  690. subprocess.call, ["qvm-prefs", target_vm_name, "label", "blue"])
  691. self._run_command_and_process_events(
  692. ["qvm-prefs", target_vm_name, "label", "red"])
  693. new_label = self._get_table_item(vm_row, "Label", Qt.DecorationRole)
  694. self.assertNotEqual(current_label.toImage(), new_label.toImage(),
  695. "Label icon did not change")
  696. icon = QIcon.fromTheme(self.qapp.domains[target_vm_name].label.icon)
  697. icon = icon.pixmap(icon_size)
  698. self.assertEqual(new_label.toImage(), icon.toImage(), "Incorrect label")
  699. def test_406_prop_change_template(self):
  700. target_vm_name = "work"
  701. vm_row = self._find_vm_row(target_vm_name)
  702. old_template = self._get_table_item(vm_row, "Template")
  703. new_template = None
  704. for vm in self.qapp.domains:
  705. if vm.klass == 'TemplateVM' and vm.name != old_template:
  706. new_template = vm.name
  707. break
  708. self.addCleanup(
  709. subprocess.call,
  710. ["qvm-prefs", target_vm_name, "template", old_template])
  711. self._run_command_and_process_events(
  712. ["qvm-prefs", target_vm_name, "template", new_template])
  713. self.assertNotEqual(old_template,
  714. self._get_table_item(vm_row, "Template"),
  715. "Template did not change")
  716. self.assertEqual(
  717. self._get_table_item(vm_row, "Template"),
  718. self.qapp.domains[target_vm_name].template.name,
  719. "Incorrect template")
  720. def test_407_prop_change_netvm(self):
  721. target_vm_name = "work"
  722. vm_row = self._find_vm_row(target_vm_name)
  723. old_netvm = self._get_table_item(vm_row, "NetVM")
  724. new_netvm = None
  725. for vm in self.qapp.domains:
  726. if getattr(vm, "provides_network", False) and vm.name != old_netvm:
  727. new_netvm = vm.name
  728. break
  729. self.addCleanup(
  730. subprocess.call, ["qvm-prefs", target_vm_name, "netvm", old_netvm])
  731. self._run_command_and_process_events(
  732. ["qvm-prefs", target_vm_name, "netvm", new_netvm])
  733. self.assertNotEqual(old_netvm,
  734. self._get_table_item(vm_row, "NetVM"),
  735. "NetVM did not change")
  736. self.assertEqual(
  737. self._get_table_item(vm_row, "NetVM"),
  738. self.qapp.domains[target_vm_name].netvm.name,
  739. "Incorrect NetVM")
  740. @unittest.expectedFailure
  741. def test_408_prop_change_internal(self):
  742. target_vm_name = "work"
  743. vm_row = self._find_vm_row(target_vm_name)
  744. self.addCleanup(subprocess.call,
  745. ["qvm-features", "--unset", "work", "interal"])
  746. self._run_command_and_process_events(
  747. ["qvm-features", "work", "interal", "1"])
  748. self.assertEqual(
  749. self._get_table_item(vm_row, "Internal"),
  750. "Yes",
  751. "Incorrect value for internal VM")
  752. self._run_command_and_process_events(
  753. ["qvm-features", "--unset", "work", "interal"])
  754. self.assertEqual(
  755. self._get_table_item(vm_row, "Internal"),
  756. "",
  757. "Incorrect value for non-internal VM")
  758. def test_409_prop_change_ip(self):
  759. target_vm_name = "work"
  760. vm_row = self._find_vm_row(target_vm_name)
  761. old_ip = self._get_table_item(vm_row, "IP")
  762. new_ip = old_ip.replace(".0.", ".5.")
  763. self.addCleanup(
  764. subprocess.call, ["qvm-prefs", target_vm_name, "ip", old_ip])
  765. self._run_command_and_process_events(
  766. ["qvm-prefs", target_vm_name, "ip", new_ip])
  767. self.assertNotEqual(old_ip,
  768. self._get_table_item(vm_row, "IP"),
  769. "IP did not change")
  770. self.assertEqual(
  771. self._get_table_item(vm_row, "IP"),
  772. self.qapp.domains[target_vm_name].ip,
  773. "Incorrect IP")
  774. def test_410_prop_change_in_backups(self):
  775. target_vm_name = "work"
  776. vm_row = self._find_vm_row(target_vm_name)
  777. old_value = self.qapp.domains[target_vm_name].include_in_backups
  778. new_value = not old_value
  779. self.addCleanup(
  780. subprocess.call,
  781. ["qvm-prefs", target_vm_name, "include_in_backups", str(old_value)])
  782. self._run_command_and_process_events(
  783. ["qvm-prefs", target_vm_name, "include_in_backups", str(new_value)])
  784. self.assertEqual(
  785. self._get_table_item(vm_row, "Internal"),
  786. "Yes" if new_value else "",
  787. "Incorrect value for include_in_backups")
  788. def test_411_prop_change_last_backup(self):
  789. target_vm_name = "work"
  790. target_timestamp = "2015-01-01 17:00:00"
  791. vm_row = self._find_vm_row(target_vm_name)
  792. old_value = self._get_table_item(vm_row, "Last backup")
  793. new_value = datetime.datetime.strptime(
  794. target_timestamp, "%Y-%m-%d %H:%M:%S")
  795. self.addCleanup(
  796. subprocess.call,
  797. ["qvm-prefs", '-D', target_vm_name, "backup_timestamp"])
  798. self._run_command_and_process_events(
  799. ["qvm-prefs", target_vm_name, "backup_timestamp",
  800. str(int(new_value.timestamp()))])
  801. self.assertNotEqual(old_value,
  802. self._get_table_item(vm_row, "Last backup"),
  803. "Last backup date did not change")
  804. self.assertEqual(
  805. self._get_table_item(vm_row, "Last backup"),
  806. target_timestamp,
  807. "Incorrect Last backup date")
  808. def test_412_prop_change_defdispvm(self):
  809. target_vm_name = "work"
  810. vm_row = self._find_vm_row(target_vm_name)
  811. old_default_dispvm =\
  812. self._get_table_item(vm_row, "Default DispVM")
  813. new_default_dispvm = None
  814. for vm in self.qapp.domains:
  815. if getattr(vm, "template_for_dispvms", False) and vm.name !=\
  816. old_default_dispvm:
  817. new_default_dispvm = vm.name
  818. break
  819. self.addCleanup(
  820. subprocess.call,
  821. ["qvm-prefs", target_vm_name, "default_dispvm", old_default_dispvm])
  822. self._run_command_and_process_events(
  823. ["qvm-prefs", target_vm_name, "default_dispvm", new_default_dispvm])
  824. self.assertNotEqual(
  825. old_default_dispvm,
  826. self._get_table_item(vm_row, "Default DispVM"),
  827. "Default DispVM did not change")
  828. self.assertEqual(
  829. self._get_table_item(vm_row, "Default DispVM"),
  830. self.qapp.domains[target_vm_name].default_dispvm.name,
  831. "Incorrect Default DispVM")
  832. def test_413_prop_change_templ_disp(self):
  833. target_vm_name = "work"
  834. vm_row = self._find_vm_row(target_vm_name)
  835. self.addCleanup(
  836. subprocess.call,
  837. ["qvm-prefs", "--default", target_vm_name, "template_for_dispvms"])
  838. self._run_command_and_process_events(
  839. ["qvm-prefs", target_vm_name, "template_for_dispvms", "True"])
  840. self.assertEqual(
  841. self._get_table_item(vm_row, "Is DVM Template"),
  842. "Yes",
  843. "Incorrect value for DVM Template")
  844. self._run_command_and_process_events(
  845. ["qvm-prefs", "--default", target_vm_name, "template_for_dispvms"])
  846. self.assertEqual(
  847. self._get_table_item(vm_row, "Is DVM Template"),
  848. "",
  849. "Incorrect value for not DVM Template")
  850. def test_414_vm_state_change(self):
  851. target_vm_name = "work"
  852. vm_row = self._find_vm_row(target_vm_name)
  853. self.assertFalse(self.qapp.domains[target_vm_name].is_running())
  854. self.addCleanup(
  855. subprocess.call,
  856. ["qvm-shutdown", "--wait", target_vm_name])
  857. self._run_command_and_process_events(
  858. ["qvm-start", target_vm_name], timeout=60)
  859. displayed_state = self._get_table_item(vm_row, "State")
  860. self.assertEqual(displayed_state['power'], 'Running',
  861. "Power state failed to update on start")
  862. self._run_command_and_process_events(
  863. ["qvm-shutdown", "--wait", target_vm_name], timeout=30)
  864. displayed_state = self._get_table_item(vm_row, "State")
  865. self.assertEqual(displayed_state['power'], 'Halted',
  866. "Power state failed to update on shutdown")
  867. def test_415_template_vm_started(self):
  868. # check whether changing state of a template_vm causes all other
  869. # vms depending on it to check theirs
  870. target_vm_name = None
  871. for vm in self.qapp.domains:
  872. if vm.klass == 'TemplateVM':
  873. for vm2 in self.qapp.domains:
  874. if getattr(vm2, 'template', None) == vm.name:
  875. target_vm_name = vm.name
  876. break
  877. if target_vm_name:
  878. break
  879. for i in range(self.dialog.table.model().rowCount()):
  880. self._get_table_vminfo(i).update = unittest.mock.Mock()
  881. self.addCleanup(
  882. subprocess.call,
  883. ["qvm-shutdown", "--wait", target_vm_name])
  884. self._run_command_and_process_events(
  885. ["qvm-start", target_vm_name], timeout=60)
  886. for i in range(self.dialog.table.model().rowCount()):
  887. call_count = self._get_table_vminfo(
  888. i).update.call_count
  889. if self._get_table_item(i, "Template") == target_vm_name:
  890. self.assertGreater(call_count, 0)
  891. elif self._get_table_item(i, "Name") == target_vm_name:
  892. self.assertGreater(call_count, 0)
  893. else:
  894. self.assertEqual(call_count, 0)
  895. def test_500_logs(self):
  896. self._select_admin_vm()
  897. self.assertTrue(self.dialog.logs_menu.isEnabled())
  898. dom0_logs = set()
  899. for c in self.dialog.logs_menu.actions():
  900. dom0_logs.add(c.text())
  901. self.assertIsNotNone(
  902. c.data(), "Empty log file found: {}".format(c.text()))
  903. self.assertIn("hypervisor", c.text(),
  904. "Log for dom0 does not contain 'hypervisor'")
  905. selected_vm = self._select_non_admin_vm(running=True).name
  906. self.assertTrue(self.dialog.logs_menu.isEnabled())
  907. vm_logs = set()
  908. for c in self.dialog.logs_menu.actions():
  909. vm_logs.add(c.text())
  910. self.assertIsNotNone(
  911. c.data(),
  912. "Empty log file found: {}".format(c.text()))
  913. self.assertIn(
  914. selected_vm,
  915. c.text(),
  916. "Log for {} does not contain its name".format(selected_vm))
  917. self.assertNotEqual(dom0_logs, vm_logs,
  918. "Same logs found for dom0 and non-adminVM")
  919. def _find_vm_row(self, vm_name):
  920. for row in range(self.dialog.table.model().rowCount()):
  921. name = self._get_table_item(row, "Name")
  922. if name == vm_name:
  923. return row
  924. return None
  925. def _count_visible_table_rows(self):
  926. result = 0
  927. for i in range(self.dialog.table.model().rowCount()):
  928. if not self.dialog.table.isRowHidden(i):
  929. result += 1
  930. return result
  931. def _run_command_and_process_events(self, command, timeout=5):
  932. """
  933. helper function to run a given command and process eventsDispatcher
  934. events
  935. :param command: list of strings, containing the command and all its
  936. parameters
  937. :param timeout: default 5 seconds
  938. :return:
  939. """
  940. asyncio.set_event_loop(self.loop)
  941. future1 = asyncio.ensure_future(self.dispatcher.listen_for_events())
  942. self.loop.run_until_complete(asyncio.sleep(0))
  943. future2 = asyncio.create_subprocess_exec(*command,
  944. stdout=subprocess.DEVNULL,
  945. stderr=subprocess.DEVNULL)
  946. (done, pending) = self.loop.run_until_complete(
  947. asyncio.wait({future1, future2}, timeout=timeout))
  948. for task in pending:
  949. with contextlib.suppress(asyncio.CancelledError):
  950. task.cancel()
  951. self.loop.call_soon(self.loop.stop)
  952. self.loop.run_forever()
  953. def _create_set_of_current_vms(self):
  954. result = set()
  955. for i in range(self.dialog.table.model().rowCount()):
  956. result.add(self._get_table_item(i, "Name"))
  957. return result
  958. def _select_admin_vm(self):
  959. for row in range(self.dialog.table.model().rowCount()):
  960. template = self._get_table_item(row, "Template")
  961. if template == 'AdminVM':
  962. index = self.dialog.table.model().index(row, 0)
  963. self.dialog.table.setCurrentIndex(index)
  964. return index.data(Qt.UserRole).vm
  965. return None
  966. def _select_non_admin_vm(self, running=None):
  967. for row in range(self.dialog.table.model().rowCount()):
  968. template = self._get_table_item(row, "Template")
  969. vm = self._get_table_vm(row)
  970. if template != 'AdminVM' and \
  971. (running is None
  972. or (running and vm.is_running())
  973. or (not running and not vm.is_running())):
  974. index = self.dialog.table.model().index(row, 0)
  975. self.dialog.table.setCurrentIndex(index)
  976. return vm
  977. return None
  978. def _select_templatevm(self, running=None):
  979. for row in range(self.dialog.table.model().rowCount()):
  980. template = self._get_table_item(row, "Template")
  981. vm = self._get_table_vm(row)
  982. if template == 'TemplateVM' and \
  983. (running is None
  984. or (running and vm.is_running())
  985. or (not running and not vm.is_running())):
  986. index = self.dialog.table.model().index(row, 0)
  987. self.dialog.table.setCurrentIndex(index)
  988. return vm
  989. return None
  990. def __check_sorting(self, column_name):
  991. last_text = None
  992. last_vm = None
  993. for row in range(self.dialog.table.model().rowCount()):
  994. vm = self._get_table_item(row, "Name")
  995. text = self._get_table_item(row, column_name)
  996. if row == 0:
  997. self.assertEqual(vm, "dom0", "dom0 is not sorted first")
  998. elif last_text is None:
  999. last_text = text
  1000. last_vm = vm
  1001. else:
  1002. if last_text == text:
  1003. self.assertGreater(
  1004. vm.lower(), last_vm.lower(),
  1005. "Incorrect sorting for {}".format(column_name))
  1006. else:
  1007. self.assertGreater(
  1008. text.lower(), last_text.lower(),
  1009. "Incorrect sorting for {}".format(column_name))
  1010. last_text = text
  1011. last_vm = vm
  1012. def _get_table_vminfo(self, row):
  1013. model = self.dialog.table.model()
  1014. return model.index(row, 0).data(Qt.UserRole)
  1015. def _get_table_vm(self, row):
  1016. model = self.dialog.table.model()
  1017. return model.index(row, 0).data(Qt.UserRole).vm
  1018. def _get_table_item(self, row, column_name, role = Qt.DisplayRole):
  1019. model = self.dialog.table.model()
  1020. column = self.dialog.qubes_model.columns_indices.index(column_name)
  1021. return model.index(row, column).data(role)
  1022. class QubeManagerThreadTest(unittest.TestCase):
  1023. def test_01_startvm_thread(self):
  1024. vm = unittest.mock.Mock(spec=['start'])
  1025. thread = qube_manager.StartVMThread(vm)
  1026. thread.run()
  1027. vm.start.assert_called_once_with()
  1028. def test_02_startvm_thread_error(self):
  1029. vm = unittest.mock.Mock(
  1030. spec=['start'],
  1031. **{'start.side_effect': exc.QubesException('Error')})
  1032. thread = qube_manager.StartVMThread(vm)
  1033. thread.run()
  1034. self.assertIsNotNone(thread.msg)
  1035. def test_10_run_command_thread(self):
  1036. vm = unittest.mock.Mock(spec=['run'])
  1037. thread = qube_manager.RunCommandThread(vm, "test_command")
  1038. thread.run()
  1039. vm.run.assert_called_once_with("test_command")
  1040. def test_11_run_command_thread_error(self):
  1041. vm = unittest.mock.Mock(spec=['run'],
  1042. **{'run.side_effect': ChildProcessError})
  1043. thread = qube_manager.RunCommandThread(vm, "test_command")
  1044. thread.run()
  1045. self.assertIsNotNone(thread.msg)
  1046. @unittest.mock.patch('subprocess.check_call')
  1047. def test_20_update_vm_thread_dom0(self, check_call):
  1048. vm = unittest.mock.Mock(spec=['klass'])
  1049. vm.klass = 'AdminVM'
  1050. thread = qube_manager.UpdateVMThread(vm)
  1051. thread.run()
  1052. check_call.assert_called_once_with(
  1053. ["/usr/bin/qubes-dom0-update", "--clean", "--gui"])
  1054. @unittest.mock.patch('builtins.open')
  1055. @unittest.mock.patch('subprocess.call')
  1056. def test_21_update_vm_thread_running(self, mock_call, mock_open):
  1057. vm = unittest.mock.Mock(
  1058. spec=['klass', 'is_running', 'run_service_for_stdio', 'run_service'],
  1059. **{'is_running.return_value': True})
  1060. vm.klass = 'AppVM'
  1061. vm.run_service_for_stdio.return_value = (b'changed=no\n', None)
  1062. thread = qube_manager.UpdateVMThread(vm)
  1063. thread.run()
  1064. mock_open.assert_called_with(
  1065. '/usr/libexec/qubes-manager/dsa-4371-update', 'rb')
  1066. vm.run_service_for_stdio.assert_called_once_with(
  1067. "qubes.VMShell", user='root', input=unittest.mock.ANY)
  1068. vm.run_service.assert_called_once_with(
  1069. "qubes.InstallUpdatesGUI", user="root", wait=False)
  1070. self.assertEqual(mock_call.call_count, 0)
  1071. @unittest.mock.patch('builtins.open')
  1072. @unittest.mock.patch('subprocess.call')
  1073. def test_22_update_vm_thread_not_running(self, mock_call, mock_open):
  1074. vm = unittest.mock.Mock(
  1075. spec=['klass', 'is_running', 'run_service_for_stdio',
  1076. 'run_service', 'start', 'name'],
  1077. **{'is_running.return_value': False})
  1078. vm.klass = 'AppVM'
  1079. vm.run_service_for_stdio.return_value = (b'changed=yes\n', None)
  1080. thread = qube_manager.UpdateVMThread(vm)
  1081. thread.run()
  1082. mock_open.assert_called_with(
  1083. '/usr/libexec/qubes-manager/dsa-4371-update', 'rb')
  1084. vm.start.assert_called_once_with()
  1085. vm.run_service_for_stdio.assert_called_once_with(
  1086. "qubes.VMShell", user='root', input=unittest.mock.ANY)
  1087. vm.run_service.assert_called_once_with(
  1088. "qubes.InstallUpdatesGUI", user="root", wait=False)
  1089. self.assertEqual(mock_call.call_count, 1)
  1090. @unittest.mock.patch('builtins.open')
  1091. @unittest.mock.patch('subprocess.check_call')
  1092. def test_23_update_vm_thread_error(self, *_args):
  1093. vm = unittest.mock.Mock(
  1094. spec=['klass', 'is_running'],
  1095. **{'is_running.side_effect': ChildProcessError})
  1096. vm.klass = 'AppVM'
  1097. thread = qube_manager.UpdateVMThread(vm)
  1098. thread.run()
  1099. self.assertIsNotNone(thread.msg)
  1100. class VMShutdownMonitorTest(unittest.TestCase):
  1101. @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
  1102. @unittest.mock.patch('PyQt5.QtCore.QTimer')
  1103. def test_01_vm_shutdown_correct(self, mock_timer, mock_question):
  1104. mock_vm = unittest.mock.Mock()
  1105. mock_vm.is_running.return_value = False
  1106. monitor = qube_manager.VmShutdownMonitor(mock_vm)
  1107. monitor.restart_vm_if_needed = unittest.mock.Mock()
  1108. monitor.check_if_vm_has_shutdown()
  1109. self.assertEqual(mock_question.call_count, 0)
  1110. self.assertEqual(mock_timer.call_count, 0)
  1111. monitor.restart_vm_if_needed.assert_called_once_with()
  1112. @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
  1113. @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
  1114. def test_02_vm_not_shutdown_wait(self, mock_timer, mock_question):
  1115. mock_question().clickedButton.return_value = 1
  1116. mock_question().addButton.return_value = 0
  1117. mock_vm = unittest.mock.Mock()
  1118. mock_vm.is_running.return_value = True
  1119. mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
  1120. monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=1)
  1121. time.sleep(3)
  1122. monitor.check_if_vm_has_shutdown()
  1123. self.assertEqual(mock_timer.call_count, 1)
  1124. @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
  1125. @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
  1126. def test_03_vm_kill(self, mock_timer, mock_question):
  1127. mock_question().clickedButton.return_value = 1
  1128. mock_question().addButton.return_value = 1
  1129. mock_vm = unittest.mock.Mock()
  1130. mock_vm.is_running.return_value = True
  1131. mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
  1132. monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=1)
  1133. time.sleep(3)
  1134. monitor.restart_vm_if_needed = unittest.mock.Mock()
  1135. monitor.check_if_vm_has_shutdown()
  1136. self.assertEqual(mock_timer.call_count, 0)
  1137. mock_vm.kill.assert_called_once_with()
  1138. monitor.restart_vm_if_needed.assert_called_once_with()
  1139. @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
  1140. @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
  1141. def test_04_check_later(self, mock_timer, mock_question):
  1142. mock_vm = unittest.mock.Mock()
  1143. mock_vm.is_running.return_value = True
  1144. mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
  1145. monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=3000)
  1146. time.sleep(1)
  1147. monitor.check_if_vm_has_shutdown()
  1148. self.assertEqual(mock_question.call_count, 0)
  1149. self.assertEqual(mock_timer.call_count, 1)
  1150. if __name__ == "__main__":
  1151. ha_syslog = logging.handlers.SysLogHandler('/dev/log')
  1152. ha_syslog.setFormatter(
  1153. logging.Formatter('%(name)s[%(process)d]: %(message)s'))
  1154. logging.root.addHandler(ha_syslog)
  1155. unittest.main()