12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475 |
- #!/usr/bin/python3
- #
- # The Qubes OS Project, https://www.qubes-os.org/
- #
- # Copyright (C) 2016 Marta Marczykowska-Górecka
- # <marmarta@invisiblethingslab.com>
- #
- # This program is free software; you can redistribute it and/or modify
- # it under the terms of the GNU General Public License as published by
- # the Free Software Foundation; either version 2 of the License, or
- # (at your option) any later version.
- #
- # This program is distributed in the hope that it will be useful,
- # but WITHOUT ANY WARRANTY; without even the implied warranty of
- # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- # GNU General Public License for more details.
- #
- # You should have received a copy of the GNU General Public License along
- # with this program; if not, write to the Free Software Foundation, Inc.,
- # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
- #
- import asyncio
- import contextlib
- import logging.handlers
- import unittest
- import unittest.mock
- import subprocess
- import datetime
- import time
- from PyQt5 import QtTest, QtCore, QtWidgets
- from PyQt5.QtCore import (Qt, QSize)
- from PyQt5.QtGui import (QIcon)
- from qubesadmin import Qubes, events, exc
- import qubesmanager.qube_manager as qube_manager
- from qubesmanager.tests import init_qtapp
- icon_size = QSize(24, 24)
- class QubeManagerTest(unittest.TestCase):
- def setUp(self):
- super(QubeManagerTest, self).setUp()
- self.qtapp, self.loop = init_qtapp()
- self.mock_qprogress = unittest.mock.patch(
- 'PyQt5.QtWidgets.QProgressDialog')
- self.mock_qprogress.start()
- self.addCleanup(self.mock_qprogress.stop)
- self.qapp = Qubes()
- self.dispatcher = events.EventsDispatcher(self.qapp)
- self.dialog = qube_manager.VmManagerWindow(
- self.qtapp, self.qapp, self.dispatcher)
- def test_000_window_loads(self):
- self.assertTrue(self.dialog.table is not None, "Window did not load")
- def test_001_correct_vms_listed(self):
- vms_in_table = []
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- self.assertIsNotNone(vm)
- vms_in_table.append(vm.name)
- # check that name is listed correctly
- name_item = self._get_table_item(row, "Name")
- self.assertEqual(name_item, vm.name,
- "Incorrect VM name for {}".format(vm.name))
- actual_vms = [vm.name for vm in self.qapp.domains]
- self.assertEqual(len(vms_in_table), len(actual_vms),
- "Incorrect number of VMs loaded")
- self.assertListEqual(sorted(vms_in_table), sorted(actual_vms),
- "Incorrect VMs loaded")
- def test_002_correct_template_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- # check that template is listed correctly
- template_item = self._get_table_item(row, "Template")
- if getattr(vm, "template", None):
- self.assertEqual(vm.template,
- template_item,
- "Incorrect template for {}".format(vm.name))
- else:
- self.assertEqual(vm.klass, template_item,
- "Incorrect class for {}".format(vm.name))
- def test_003_correct_netvm_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- # check that netvm is listed correctly
- netvm_item = self._get_table_item(row, "NetVM")
- netvm_value = getattr(vm, "netvm", None)
- if not netvm_value:
- netvm_value = "n/a"
- if netvm_value and hasattr(vm, "netvm") \
- and vm.property_is_default("netvm"):
- netvm_value = "default ({})".format(netvm_value)
- self.assertEqual(netvm_value,
- netvm_item,
- "Incorrect netvm for {}".format(vm.name))
- def test_004_correct_disk_usage_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- size_item = self._get_table_item(row, "Disk Usage")
- if vm.klass == 'AdminVM':
- size_value = "n/a"
- else:
- size_value = round(vm.get_disk_utilization() / (1024 * 1024), 2)
- size_value = str(size_value) + " MiB"
- self.assertEqual(size_value,
- size_item,
- "Incorrect size for {}".format(vm.name))
- def test_005_correct_internal_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- internal_item = self._get_table_item(row, "Internal")
- internal_value = "Yes" if vm.features.get('internal', False) else ""
- self.assertEqual(internal_item, internal_value,
- "Incorrect internal value for {}".format(vm.name))
- def test_006_correct_ip_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- ip_item = self._get_table_item(row, "IP")
- if hasattr(vm, 'ip'):
- ip_value = getattr(vm, 'ip')
- ip_value = "" if ip_value is None else ip_value
- else:
- ip_value = "n/a"
- self.assertEqual(ip_value, ip_item,
- "Incorrect ip value for {}".format(vm.name))
- def test_007_incl_in_backups_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- incl_backups_item = self._get_table_item(row, "Include in backups")
- incl_backups_value = getattr(vm, 'include_in_backups', False)
- incl_backups_value = "Yes" if incl_backups_value else ""
- self.assertEqual(
- incl_backups_value, incl_backups_item,
- "Incorrect include in backups value for {}".format(vm.name))
- def test_008_last_backup_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- last_backup_item = self._get_table_item(row, "Last backup")
- last_backup_value = getattr(vm, 'backup_timestamp', None)
- if last_backup_value:
- last_backup_value = str(
- datetime.datetime.fromtimestamp(last_backup_value))
- self.assertEqual(
- last_backup_value, last_backup_item,
- "Incorrect last backup value for {}".format(vm.name))
- def test_009_def_dispvm_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- def_dispvm_item = self._get_table_item(row, "Default DispVM")
- if vm.property_is_default("default_dispvm"):
- def_dispvm_value = "default ({})".format(
- self.qapp.default_dispvm)
- else:
- def_dispvm_value = getattr(vm, "default_dispvm", None)
- self.assertEqual(
- def_dispvm_value, def_dispvm_item,
- "Incorrect default dispvm value for {}".format(vm.name))
- def test_010_is_dvm_template_listed(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- is_dvm_template_item = self._get_table_item(row, "Is DVM Template")
- is_dvm_template_value = "Yes" if \
- getattr(vm, "template_for_dispvms", False) else ""
- self.assertEqual(
- is_dvm_template_value, is_dvm_template_item,
- "Incorrect is DVM template value for {}".format(vm.name))
- def test_011_is_label_correct(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- icon = QIcon.fromTheme(vm.label.icon)
- icon = icon.pixmap(icon_size)
- label_pixmap = self._get_table_item(row, "Label", Qt.DecorationRole)
- self.assertEqual(label_pixmap.toImage(), icon.toImage())
- def test_012_is_state_correct(self):
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_vm(row)
- displayed_power_state = self._get_table_item(row, "State")['power']
- self.assertEqual(
- displayed_power_state, vm.get_power_state(),
- "Wrong power state displayed for {}".format(vm.name))
- def test_013_incorrect_settings_file(self):
- mock_settings = unittest.mock.MagicMock(spec=QtCore.QSettings)
- settings_result_dict = {"view/sort_column": "Cthulhu",
- "view/sort_order": "Fhtagn",
- "view/menubar_visible": "R'lyeh"
- }
- mock_settings.side_effect = (
- lambda x, *args, **kwargs: settings_result_dict.get(x))
- with unittest.mock.patch('PyQt5.QtCore.QSettings.value',
- mock_settings),\
- unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')\
- as mock_warning:
- self.dialog = qube_manager.VmManagerWindow(
- self.qtapp, self.qapp, self.dispatcher)
- self.assertEqual(mock_warning.call_count, 1)
- def test_100_sorting(self):
- col = self.dialog.qubes_model.columns_indices.index("Template")
- self.dialog.table.sortByColumn(col, QtCore.Qt.AscendingOrder)
- self.__check_sorting("Template")
- col = self.dialog.qubes_model.columns_indices.index("Name")
- self.dialog.table.sortByColumn(col, QtCore.Qt.AscendingOrder)
- self.__check_sorting("Name")
- @unittest.mock.patch('qubesmanager.qube_manager.QSettings.setValue')
- def test_101_hide_column(self, mock_settings):
- model = self.dialog.qubes_model
- action_no = model.columns_indices.index('Is DVM Template')
- self.dialog.menu_view.actions()[action_no].trigger()
- mock_settings.assert_called_with('columns/Is DVM Template', True)
- self.dialog.menu_view.actions()[action_no].trigger()
- mock_settings.assert_called_with('columns/Is DVM Template', False)
- @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
- def test_200_vm_open_settings(self, mock_window):
- selected_vm = self._select_non_admin_vm()
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_settings)
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_window.assert_called_once_with(
- selected_vm, qapp=self.qtapp, init_page="basic", qubesapp=self.qapp)
- def test_201_vm_open_settings_admin(self):
- self._select_admin_vm()
- self.assertFalse(self.dialog.action_settings.isEnabled(),
- "Settings not disabled for admin VM")
- self.assertFalse(self.dialog.action_editfwrules.isEnabled(),
- "Editfw not disabled for admin VM")
- self.assertFalse(self.dialog.action_appmenus.isEnabled(),
- "Appmenus not disabled for admin VM")
- @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
- def test_202_vm_open_firewall(self, mock_window):
- selected_vm = self._select_non_admin_vm()
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_editfwrules)
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_window.assert_called_once_with(
- selected_vm, qapp=self.qtapp, init_page="firewall",
- qubesapp=self.qapp)
- @unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')
- def test_203_vm_open_apps(self, mock_window):
- selected_vm = self._select_non_admin_vm()
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_appmenus)
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_window.assert_called_once_with(
- selected_vm, qapp=self.qtapp, init_page="applications",
- qubesapp=self.qapp)
- def test_204_vm_keyboard(self):
- selected_vm = self._select_non_admin_vm(running=True)
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_set_keyboard_layout)
- with unittest.mock.patch.object(selected_vm, 'run') as mock_run:
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_run.assert_called_once_with("qubes-change-keyboard-layout")
- def test_205_vm_keyboard_not_running(self):
- selected_vm = self._select_non_admin_vm(running=False)
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_set_keyboard_layout)
- with unittest.mock.patch.object(selected_vm, 'run') as mock_run:
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- self.assertEqual(mock_run.call_count, 0,
- "Keyboard change called on a halted VM")
- def test_206_dom0_keyboard(self):
- self._select_admin_vm()
- self.assertFalse(self.dialog.action_set_keyboard_layout.isEnabled())
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- def test_207_update_vm_not_running(self, _):
- selected_vm = self._select_templatevm(running=False)
- self.assertIsNotNone(selected_vm, "No valid template VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_updatevm)
- with unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread') \
- as mock_update:
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_update.assert_called_once_with(selected_vm)
- mock_update().start.assert_called_once_with()
- def test_208_update_vm_admin(self):
- selected_vm = self._select_admin_vm()
- self.assertIsNotNone(selected_vm, "No valid admin VM found")
- widget = self.dialog.toolbar.widgetForAction(
- self.dialog.action_updatevm)
- with unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread') \
- as mock_update:
- QtTest.QTest.mouseClick(widget,
- QtCore.Qt.LeftButton)
- mock_update.assert_called_once_with(selected_vm)
- mock_update().start.assert_called_once_with()
- @unittest.mock.patch("PyQt5.QtWidgets.QInputDialog.getText",
- return_value=("command to run", True))
- def test_209_run_command_in_vm(self, _):
- selected_vm = self._select_non_admin_vm()
- self.assertIsNotNone(selected_vm, "No valid non-admin VM found")
- with unittest.mock.patch('qubesmanager.qube_manager.RunCommandThread') \
- as mock_thread:
- self.dialog.action_run_command_in_vm.trigger()
- mock_thread.assert_called_once_with(selected_vm, "command to run")
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- def test_210_run_command_in_adminvm(self):
- self._select_admin_vm()
- self.assertFalse(self.dialog.action_run_command_in_vm.isEnabled(),
- "Should not be able to run commands for dom0")
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.warning")
- def test_211_pausevm(self, mock_warn):
- selected_vm = self._select_non_admin_vm(running=True)
- self.assertTrue(self.dialog.action_pausevm.isEnabled(),
- "Pause not enabled for a running VM")
- with unittest.mock.patch.object(selected_vm, 'pause') as mock_pause:
- self.dialog.action_pausevm.trigger()
- mock_pause.assert_called_once_with()
- mock_pause.side_effect = exc.QubesException('Error')
- self.dialog.action_pausevm.trigger()
- self.assertEqual(mock_warn.call_count, 1)
- def test_212_resumevm(self):
- selected_vm = self._select_non_admin_vm(running=False)
- with unittest.mock.patch.object(selected_vm, 'get_power_state')\
- as mock_state, \
- unittest.mock.patch.object(selected_vm, 'unpause')\
- as mock_unpause:
- mock_state.return_value = 'Paused'
- self.dialog.action_resumevm.trigger()
- mock_unpause.assert_called_once_with()
- with unittest.mock.patch('qubesmanager.qube_manager.StartVMThread') \
- as mock_thread:
- self.dialog.action_resumevm.trigger()
- mock_thread.assert_called_once_with(selected_vm)
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- def test_213_resume_running_vm(self):
- self._select_non_admin_vm(running=True)
- self.assertFalse(self.dialog.action_resumevm.isEnabled())
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
- @unittest.mock.patch('qubesmanager.qube_manager.VmShutdownMonitor')
- def test_214_shutdownvm(self, mock_monitor, mock_timer, _):
- selected_vm = self._select_non_admin_vm(running=True)
- with unittest.mock.patch.object(selected_vm, 'shutdown')\
- as mock_shutdown:
- self.dialog.action_shutdownvm.trigger()
- mock_shutdown.assert_called_once_with()
- mock_monitor.assert_called_once_with(
- selected_vm,
- unittest.mock.ANY, unittest.mock.ANY,
- unittest.mock.ANY, unittest.mock.ANY)
- mock_timer.assert_called_once_with(unittest.mock.ANY,
- unittest.mock.ANY)
- def test_215_shutdown_halted_vm(self):
- self._select_non_admin_vm(running=False)
- self.assertFalse(self.dialog.action_shutdownvm.isEnabled())
- @unittest.mock.patch('qubesmanager.create_new_vm.NewVmDlg')
- def test_216_create_vm(self, mock_new_vm):
- action = self.dialog.action_createvm
- self.assertTrue(action.isEnabled())
- action.trigger()
- self.assertEqual(mock_new_vm.call_count, 1,
- "Create New VM window did not appear")
- def test_217_remove_admin_vm(self):
- self._select_admin_vm()
- self.assertFalse(self.dialog.action_removevm.isEnabled())
- @unittest.mock.patch("qubesmanager.qube_manager.QMessageBox")
- @unittest.mock.patch('qubesadmin.utils.vm_dependencies')
- def test_218_remove_vm_dependencies(self, mock_dependencies, mock_msgbox):
- mock_vm = unittest.mock.Mock(spec=['name'],
- **{'name.return_value': 'test-vm'})
- mock_dependencies.return_value = [(mock_vm, "test_prop")]
- action = self.dialog.action_removevm
- self._select_non_admin_vm()
- action.trigger()
- mock_msgbox().show.assert_called_with()
- @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
- @unittest.mock.patch("PyQt5.QtWidgets.QInputDialog.getText")
- @unittest.mock.patch('qubesadmin.utils.vm_dependencies')
- def test_219_remove_vm_no_depencies(
- self, mock_dependencies, mock_input, mock_warning):
- action = self.dialog.action_removevm
- selected_vm = self._select_non_admin_vm(running=False)
- # test with no dependencies
- mock_dependencies.return_value = None
- with unittest.mock.patch('qubesmanager.common_threads.RemoveVMThread')\
- as mock_thread:
- mock_input.return_value = (selected_vm, False)
- action.trigger()
- self.assertEqual(mock_thread.call_count, 0,
- "VM removed despite user clicking 'cancel")
- mock_input.return_value = ("wrong_name", True)
- action.trigger()
- self.assertEqual(mock_warning.call_count, 1)
- self.assertEqual(mock_thread.call_count, 0,
- "VM removed despite user not confirming the name")
- mock_input.return_value = (selected_vm.name, True)
- action.trigger()
- mock_thread.assert_called_once_with(selected_vm)
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- def test_220_restartvm_halted_vm(self):
- self._select_non_admin_vm(running=False)
- self.assertFalse(self.dialog.action_restartvm.isEnabled())
- @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
- @unittest.mock.patch('qubesmanager.qube_manager.VmShutdownMonitor')
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- def test_221_restartvm_running_vm(self, _msgbox, mock_monitor, _qtimer):
- selected_vm = self._select_non_admin_vm(running=True)
- action = self.dialog.action_restartvm
- # currently the VM is running
- with unittest.mock.patch.object(selected_vm, 'shutdown')\
- as mock_shutdown:
- action.trigger()
- mock_shutdown.assert_called_once_with()
- mock_monitor.assert_called_once_with(
- selected_vm, unittest.mock.ANY,
- unittest.mock.ANY, True, unittest.mock.ANY)
- @unittest.mock.patch('qubesmanager.qube_manager.StartVMThread')
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- def test_222_restartvm_shutdown_meantime(self, _, mock_thread):
- selected_vm = self._select_non_admin_vm(running=True)
- action = self.dialog.action_restartvm
- # it was shutdown in the meantime
- with unittest.mock.patch.object(
- selected_vm, 'is_running', **{'return_value': False}):
- action.trigger()
- mock_thread.assert_called_once_with(selected_vm)
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- @unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread')
- def test_223_updatevm_running(self, mock_thread):
- selected_vm = self._select_non_admin_vm(running=True)
- self.dialog.action_updatevm.trigger()
- mock_thread.assert_called_once_with(selected_vm)
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- @unittest.mock.patch('qubesmanager.qube_manager.UpdateVMThread')
- def test_224_updatevm_halted(self, mock_thread, _):
- selected_vm = self._select_non_admin_vm(running=False)
- self.dialog.action_updatevm.trigger()
- mock_thread.assert_called_once_with(selected_vm)
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Yes)
- def test_224_killvm(self, _):
- selected_vm = self._select_non_admin_vm(running=True)
- action = self.dialog.action_killvm
- with unittest.mock.patch.object(selected_vm, 'kill') as mock_kill:
- action.trigger()
- mock_kill.assert_called_once_with()
- @unittest.mock.patch("PyQt5.QtWidgets.QMessageBox.question",
- return_value=QtWidgets.QMessageBox.Cancel)
- def test_225_killvm_cancel(self, _):
- selected_vm = self._select_non_admin_vm(running=True)
- action = self.dialog.action_killvm
- with unittest.mock.patch.object(selected_vm, 'kill') as mock_kill:
- action.trigger()
- self.assertEqual(mock_kill.call_count, 0,
- "Ignored Cancel on kill VM")
- @unittest.mock.patch('qubesmanager.global_settings.GlobalSettingsWindow')
- def test_226_global_settings(self, mock_settings):
- self._select_non_admin_vm()
- self.dialog.action_global_settings.trigger()
- self.assertEqual(mock_settings.call_count, 1,
- "Global Settings not opened")
- self._select_admin_vm()
- self.dialog.action_global_settings.trigger()
- self.assertEqual(mock_settings.call_count, 2,
- "Global Settings not opened for the second time")
- @unittest.mock.patch('qubesmanager.backup.BackupVMsWindow')
- def test_227_backup(self, mock_backup):
- self.dialog.action_backup.trigger()
- self.assertTrue(self.dialog.action_backup.isEnabled())
- self.assertEqual(mock_backup.call_count, 1,
- "Backup window does not appear")
- @unittest.mock.patch('qubesmanager.restore.RestoreVMsWindow')
- def test_228_restore(self, mock_restore):
- self.dialog.action_restore.trigger()
- self.assertTrue(self.dialog.action_restore.isEnabled())
- self.assertEqual(mock_restore.call_count, 1,
- "Backup window does not appear")
- @unittest.mock.patch('qubesmanager.qube_manager.AboutDialog')
- def test_229_about_qubes(self, mock_about):
- self.assertTrue(self.dialog.action_about_qubes.isEnabled())
- self.dialog.action_about_qubes.trigger()
- self.assertEqual(
- mock_about.call_count, 1, "About window does not appear")
- def test_230_exit_action(self):
- self.assertTrue(self.dialog.action_exit.isEnabled())
- with unittest.mock.patch.object(self.dialog, 'close') as mock_close:
- self.dialog.action_exit.trigger()
- mock_close.assert_called_once_with()
- @unittest.mock.patch('subprocess.check_call')
- def test_231_template_manager(self, mock_subprocess):
- self.assertTrue(self.dialog.action_manage_templates.isEnabled())
- self.dialog.action_manage_templates.trigger()
- mock_subprocess.assert_called_once_with('qubes-template-manager')
- @unittest.mock.patch('qubesmanager.common_threads.CloneVMThread')
- @unittest.mock.patch('PyQt5.QtWidgets.QInputDialog.getText')
- def test_232_clonevm(self, mock_input, mock_thread):
- action = self.dialog.action_clonevm
- self._select_admin_vm()
- self.assertFalse(action.isEnabled())
- selected_vm = self._select_non_admin_vm()
- self.assertTrue(action.isEnabled())
- mock_input.return_value = (selected_vm.name + "clone1", False)
- action.trigger()
- self.assertEqual(mock_thread.call_count, 0,
- "Ignores cancelling clone VM")
- mock_input.return_value = (selected_vm.name + "clone1", True)
- action.trigger()
- mock_thread.assert_called_once_with(selected_vm,
- selected_vm.name + "clone1")
- mock_thread().finished.connect.assert_called_once_with(
- self.dialog.clear_threads)
- mock_thread().start.assert_called_once_with()
- def test_233_search_action(self):
- self.qtapp.setActiveWindow(self.dialog.searchbox)
- self.dialog.action_search.trigger()
- self.assertTrue(self.dialog.searchbox.hasFocus())
- # input text
- self.dialog.searchbox.setText("sys")
- # click outside the widget
- QtTest.QTest.mouseClick(self.dialog.table, QtCore.Qt.LeftButton)
- # click the widget, check if it is correctly activated and the whole
- # text was selected
- QtTest.QTest.mouseClick(self.dialog.searchbox, QtCore.Qt.LeftButton)
- self.assertTrue(self.dialog.searchbox.hasFocus())
- self.assertEqual(self.dialog.searchbox.selectedText(), "sys")
- def test_234_searchbox(self):
- # look for sys
- self.dialog.searchbox.setText("sys")
- expected_number = \
- len([vm for vm in self.qapp.domains if "sys" in vm.name])
- actual_number = self._count_visible_table_rows()
- self.assertEqual(expected_number, actual_number,
- "Incorrect number of vms shown for 'sys'")
- # clear search
- self.dialog.searchbox.setText("")
- expected_number = len([vm for vm in self.qapp.domains])
- actual_number = self._count_visible_table_rows()
- self.assertEqual(expected_number, actual_number,
- "Incorrect number of vms shown for cleared search box")
- def test_235_hide_show_toolbars(self):
- with unittest.mock.patch('PyQt5.QtCore.QSettings.setValue')\
- as mock_setvalue:
- self.dialog.action_menubar.trigger()
- mock_setvalue.assert_called_with('view/menubar_visible', False)
- self.dialog.action_toolbar.trigger()
- mock_setvalue.assert_called_with('view/toolbar_visible', False)
- self.assertFalse(self.dialog.menubar.isVisible(),
- "Menubar not hidden correctly")
- self.assertFalse(self.dialog.toolbar.isVisible(),
- "Toolbar not hidden correctly")
- def test_236_clear_searchbox(self):
- self.dialog.searchbox.setText("text")
- self.assertEqual(self.dialog.searchbox.text(), "text")
- QtTest.QTest.keyPress(self.dialog, QtCore.Qt.Key_Escape)
- self.assertEqual(self.dialog.searchbox.text(), "",
- "Escape failed to clear searchbox")
- expected_number = len([vm for vm in self.qapp.domains])
- actual_number = self._count_visible_table_rows()
- self.assertEqual(expected_number, actual_number,
- "Incorrect number of vms shown for cleared search box")
- @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.information')
- @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
- def test_300_clear_threads(self, mock_warning, mock_info):
- mock_thread_finished_ok = unittest.mock.Mock(
- spec=['isFinished', 'msg', 'msg_is_success'],
- msg=None, msg_is_success=False,
- **{'isFinished.return_value': True})
- mock_thread_not_finished = unittest.mock.Mock(
- spec=['isFinished', 'msg', 'msg_is_success'],
- msg=None, msg_is_success=False,
- **{'isFinished.return_value': False})
- mock_thread_finished_error = unittest.mock.Mock(
- spec=['isFinished', 'msg', 'msg_is_success'],
- msg=("Error", "Error"), msg_is_success=False,
- **{'isFinished.return_value': True})
- mock_thread_fin_error_success = unittest.mock.Mock(
- spec=['isFinished', 'msg', 'msg_is_success'],
- msg=("Done", "Done"), msg_is_success=True,
- **{'isFinished.return_value': True})
- # single finished thread
- self.dialog.threads_list = [mock_thread_not_finished,
- mock_thread_finished_ok]
- self.dialog.clear_threads()
- self.assertEqual(mock_warning.call_count, 0)
- self.assertEqual(mock_info.call_count, 0)
- self.assertEqual(len(self.dialog.threads_list), 1)
- # an error thread and some in-progress ones
- self.dialog.threads_list = [mock_thread_not_finished,
- mock_thread_not_finished,
- mock_thread_finished_error]
- self.dialog.clear_threads()
- self.assertEqual(mock_warning.call_count, 1)
- self.assertEqual(mock_info.call_count, 0)
- self.assertEqual(len(self.dialog.threads_list), 2)
- # an error-success thread and some in-progress ones
- self.dialog.threads_list = [mock_thread_not_finished,
- mock_thread_not_finished,
- mock_thread_fin_error_success,
- mock_thread_finished_error]
- self.dialog.clear_threads()
- self.assertEqual(mock_warning.call_count, 1)
- self.assertEqual(mock_info.call_count, 1)
- self.assertEqual(len(self.dialog.threads_list), 3)
- def test_400_event_domain_added(self):
- number_of_vms = self.dialog.table.model().rowCount()
- self.addCleanup(subprocess.call, ["qvm-remove", "-f", "test-vm"])
- self._run_command_and_process_events(
- ["qvm-create", "--label", "red", "test-vm"])
- # a single row was added to the table
- self.assertEqual(self.dialog.table.model().rowCount(), number_of_vms+1)
- # table contains the correct vms
- vms_in_table = self._create_set_of_current_vms()
- vms_in_system = set([vm.name for vm in self.qapp.domains])
- self.assertEqual(vms_in_table, vms_in_system, "Table not updated "
- "correctly after add")
- # check if sorting works
- self.__check_sorting("Name")
- # try opening settings for the added vm
- for row in range(self.dialog.table.model().rowCount()):
- name = self._get_table_item(row, "Name")
- if name == "test-vm":
- index = self.dialog.table.model().index(row, 0)
- self.dialog.table.setCurrentIndex(index)
- break
- with unittest.mock.patch('qubesmanager.settings.VMSettingsWindow')\
- as mock_settings:
- self.dialog.action_settings.trigger()
- mock_settings.assert_called_once_with(
- self.qapp.domains["test-vm"], qapp = self.qtapp,
- init_page = "basic", qubesapp = self.qapp)
- def test_401_event_domain_removed(self):
- initial_vms = self._create_set_of_current_vms()
- self._run_command_and_process_events(
- ["qvm-create", "--label", "red", "test-vm"])
- current_vms = self._create_set_of_current_vms()
- self.assertEqual(len(initial_vms) + 1, len(current_vms))
- self._run_command_and_process_events(
- ["qvm-remove", "--force", "test-vm"])
- current_vms = self._create_set_of_current_vms()
- self.assertEqual(initial_vms, current_vms)
- # check if sorting works
- self.__check_sorting("Name")
- def test_403_event_dispvm_added(self):
- initial_vms = self._create_set_of_current_vms()
- dispvm_template = None
- for vm in self.qapp.domains:
- if getattr(vm, "template_for_dispvms", False):
- dispvm_template = vm.name
- break
- self.assertIsNotNone(dispvm_template,
- "Cannot find a template for dispVMs")
- # this requires very long timeout, because it takes time for the
- # dispvm to vanish
- self._run_command_and_process_events(
- ["qvm-run", "--dispvm", dispvm_template, "true"], timeout=60)
- final_vms = self._create_set_of_current_vms()
- self.assertEqual(initial_vms, final_vms,
- "Failed handling of a created-and-removed dispvm")
- def test_404_crashing_dispvm(self):
- initial_vms = self._create_set_of_current_vms()
- dispvm_template = None
- for vm in self.qapp.domains:
- if getattr(vm, "template_for_dispvms", False):
- dispvm_template = vm.name
- break
- self.assertIsNotNone(dispvm_template,
- "Cannot find a template for dispVMs")
- current_memory = getattr(self.qapp.domains[dispvm_template], "memory")
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", dispvm_template, "memory", str(current_memory)])
- subprocess.check_call(
- ["qvm-prefs", dispvm_template, "memory", "600000"])
- self._run_command_and_process_events(
- ["qvm-run", "--dispvm", dispvm_template, "true"], timeout=30)
- final_vms = self._create_set_of_current_vms()
- self.assertEqual(initial_vms, final_vms,
- "Failed handling of dispvm that crashed on start")
- def test_405_prop_change_label(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- current_label = self._get_table_item(vm_row, "Label", Qt.DecorationRole)
- self.addCleanup(
- subprocess.call, ["qvm-prefs", target_vm_name, "label", "blue"])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "label", "red"])
- new_label = self._get_table_item(vm_row, "Label", Qt.DecorationRole)
- self.assertNotEqual(current_label.toImage(), new_label.toImage(),
- "Label icon did not change")
- icon = QIcon.fromTheme(self.qapp.domains[target_vm_name].label.icon)
- icon = icon.pixmap(icon_size)
- self.assertEqual(new_label.toImage(), icon.toImage(), "Incorrect label")
- def test_406_prop_change_template(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- old_template = self._get_table_item(vm_row, "Template")
- new_template = None
- for vm in self.qapp.domains:
- if vm.klass == 'TemplateVM' and vm.name != old_template:
- new_template = vm.name
- break
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", target_vm_name, "template", old_template])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "template", new_template])
- self.assertNotEqual(old_template,
- self._get_table_item(vm_row, "Template"),
- "Template did not change")
- self.assertEqual(
- self._get_table_item(vm_row, "Template"),
- self.qapp.domains[target_vm_name].template.name,
- "Incorrect template")
- def test_407_prop_change_netvm(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- old_netvm = self._get_table_item(vm_row, "NetVM")
- new_netvm = None
- for vm in self.qapp.domains:
- if getattr(vm, "provides_network", False) and vm.name != old_netvm:
- new_netvm = vm.name
- break
- self.addCleanup(
- subprocess.call, ["qvm-prefs", target_vm_name, "netvm", old_netvm])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "netvm", new_netvm])
- self.assertNotEqual(old_netvm,
- self._get_table_item(vm_row, "NetVM"),
- "NetVM did not change")
- self.assertEqual(
- self._get_table_item(vm_row, "NetVM"),
- self.qapp.domains[target_vm_name].netvm.name,
- "Incorrect NetVM")
- @unittest.expectedFailure
- def test_408_prop_change_internal(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- self.addCleanup(subprocess.call,
- ["qvm-features", "--unset", "work", "interal"])
- self._run_command_and_process_events(
- ["qvm-features", "work", "interal", "1"])
- self.assertEqual(
- self._get_table_item(vm_row, "Internal"),
- "Yes",
- "Incorrect value for internal VM")
- self._run_command_and_process_events(
- ["qvm-features", "--unset", "work", "interal"])
- self.assertEqual(
- self._get_table_item(vm_row, "Internal"),
- "",
- "Incorrect value for non-internal VM")
- def test_409_prop_change_ip(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- old_ip = self._get_table_item(vm_row, "IP")
- new_ip = old_ip.replace(".0.", ".5.")
- self.addCleanup(
- subprocess.call, ["qvm-prefs", target_vm_name, "ip", old_ip])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "ip", new_ip])
- self.assertNotEqual(old_ip,
- self._get_table_item(vm_row, "IP"),
- "IP did not change")
- self.assertEqual(
- self._get_table_item(vm_row, "IP"),
- self.qapp.domains[target_vm_name].ip,
- "Incorrect IP")
- def test_410_prop_change_in_backups(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- old_value = self.qapp.domains[target_vm_name].include_in_backups
- new_value = not old_value
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", target_vm_name, "include_in_backups", str(old_value)])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "include_in_backups", str(new_value)])
- self.assertEqual(
- self._get_table_item(vm_row, "Internal"),
- "Yes" if new_value else "",
- "Incorrect value for include_in_backups")
- def test_411_prop_change_last_backup(self):
- target_vm_name = "work"
- target_timestamp = "2015-01-01 17:00:00"
- vm_row = self._find_vm_row(target_vm_name)
- old_value = self._get_table_item(vm_row, "Last backup")
- new_value = datetime.datetime.strptime(
- target_timestamp, "%Y-%m-%d %H:%M:%S")
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", '-D', target_vm_name, "backup_timestamp"])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "backup_timestamp",
- str(int(new_value.timestamp()))])
- self.assertNotEqual(old_value,
- self._get_table_item(vm_row, "Last backup"),
- "Last backup date did not change")
- self.assertEqual(
- self._get_table_item(vm_row, "Last backup"),
- target_timestamp,
- "Incorrect Last backup date")
- def test_412_prop_change_defdispvm(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- old_default_dispvm =\
- self._get_table_item(vm_row, "Default DispVM")
- new_default_dispvm = None
- for vm in self.qapp.domains:
- if getattr(vm, "template_for_dispvms", False) and vm.name !=\
- old_default_dispvm:
- new_default_dispvm = vm.name
- break
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", target_vm_name, "default_dispvm", old_default_dispvm])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "default_dispvm", new_default_dispvm])
- self.assertNotEqual(
- old_default_dispvm,
- self._get_table_item(vm_row, "Default DispVM"),
- "Default DispVM did not change")
- self.assertEqual(
- self._get_table_item(vm_row, "Default DispVM"),
- self.qapp.domains[target_vm_name].default_dispvm.name,
- "Incorrect Default DispVM")
- def test_413_prop_change_templ_disp(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- self.addCleanup(
- subprocess.call,
- ["qvm-prefs", "--default", target_vm_name, "template_for_dispvms"])
- self._run_command_and_process_events(
- ["qvm-prefs", target_vm_name, "template_for_dispvms", "True"])
- self.assertEqual(
- self._get_table_item(vm_row, "Is DVM Template"),
- "Yes",
- "Incorrect value for DVM Template")
- self._run_command_and_process_events(
- ["qvm-prefs", "--default", target_vm_name, "template_for_dispvms"])
- self.assertEqual(
- self._get_table_item(vm_row, "Is DVM Template"),
- "",
- "Incorrect value for not DVM Template")
- def test_414_vm_state_change(self):
- target_vm_name = "work"
- vm_row = self._find_vm_row(target_vm_name)
- self.assertFalse(self.qapp.domains[target_vm_name].is_running())
- self.addCleanup(
- subprocess.call,
- ["qvm-shutdown", "--wait", target_vm_name])
- self._run_command_and_process_events(
- ["qvm-start", target_vm_name], timeout=60)
- displayed_state = self._get_table_item(vm_row, "State")
- self.assertEqual(displayed_state['power'], 'Running',
- "Power state failed to update on start")
- self._run_command_and_process_events(
- ["qvm-shutdown", "--wait", target_vm_name], timeout=30)
- displayed_state = self._get_table_item(vm_row, "State")
- self.assertEqual(displayed_state['power'], 'Halted',
- "Power state failed to update on shutdown")
- def test_415_template_vm_started(self):
- # check whether changing state of a template_vm causes all other
- # vms depending on it to check theirs
- target_vm_name = None
- for vm in self.qapp.domains:
- if vm.klass == 'TemplateVM':
- for vm2 in self.qapp.domains:
- if getattr(vm2, 'template', None) == vm.name:
- target_vm_name = vm.name
- break
- if target_vm_name:
- break
- for i in range(self.dialog.table.model().rowCount()):
- self._get_table_vminfo(i).update = unittest.mock.Mock()
- self.addCleanup(
- subprocess.call,
- ["qvm-shutdown", "--wait", target_vm_name])
- self._run_command_and_process_events(
- ["qvm-start", target_vm_name], timeout=60)
- for i in range(self.dialog.table.model().rowCount()):
- call_count = self._get_table_vminfo(
- i).update.call_count
- if self._get_table_item(i, "Template") == target_vm_name:
- self.assertGreater(call_count, 0)
- elif self._get_table_item(i, "Name") == target_vm_name:
- self.assertGreater(call_count, 0)
- else:
- self.assertEqual(call_count, 0)
- def test_500_logs(self):
- self._select_admin_vm()
- self.assertTrue(self.dialog.logs_menu.isEnabled())
- dom0_logs = set()
- for c in self.dialog.logs_menu.actions():
- dom0_logs.add(c.text())
- self.assertIsNotNone(
- c.data(), "Empty log file found: {}".format(c.text()))
- self.assertIn("hypervisor", c.text(),
- "Log for dom0 does not contain 'hypervisor'")
- selected_vm = self._select_non_admin_vm(running=True).name
- self.assertTrue(self.dialog.logs_menu.isEnabled())
- vm_logs = set()
- for c in self.dialog.logs_menu.actions():
- vm_logs.add(c.text())
- self.assertIsNotNone(
- c.data(),
- "Empty log file found: {}".format(c.text()))
- self.assertIn(
- selected_vm,
- c.text(),
- "Log for {} does not contain its name".format(selected_vm))
- self.assertNotEqual(dom0_logs, vm_logs,
- "Same logs found for dom0 and non-adminVM")
- def _find_vm_row(self, vm_name):
- for row in range(self.dialog.table.model().rowCount()):
- name = self._get_table_item(row, "Name")
- if name == vm_name:
- return row
- return None
- def _count_visible_table_rows(self):
- result = 0
- for i in range(self.dialog.table.model().rowCount()):
- if not self.dialog.table.isRowHidden(i):
- result += 1
- return result
- def _run_command_and_process_events(self, command, timeout=5):
- """
- helper function to run a given command and process eventsDispatcher
- events
- :param command: list of strings, containing the command and all its
- parameters
- :param timeout: default 5 seconds
- :return:
- """
- asyncio.set_event_loop(self.loop)
- future1 = asyncio.ensure_future(self.dispatcher.listen_for_events())
- self.loop.run_until_complete(asyncio.sleep(0))
- future2 = asyncio.create_subprocess_exec(*command,
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL)
- (done, pending) = self.loop.run_until_complete(
- asyncio.wait({future1, future2}, timeout=timeout))
- for task in pending:
- with contextlib.suppress(asyncio.CancelledError):
- task.cancel()
- self.loop.call_soon(self.loop.stop)
- self.loop.run_forever()
- def _create_set_of_current_vms(self):
- result = set()
- for i in range(self.dialog.table.model().rowCount()):
- result.add(self._get_table_item(i, "Name"))
- return result
- def _select_admin_vm(self):
- for row in range(self.dialog.table.model().rowCount()):
- template = self._get_table_item(row, "Template")
- if template == 'AdminVM':
- index = self.dialog.table.model().index(row, 0)
- self.dialog.table.setCurrentIndex(index)
- return index.data(Qt.UserRole).vm
- return None
- def _select_non_admin_vm(self, running=None):
- for row in range(self.dialog.table.model().rowCount()):
- template = self._get_table_item(row, "Template")
- vm = self._get_table_vm(row)
- if template != 'AdminVM' and \
- (running is None
- or (running and vm.is_running())
- or (not running and not vm.is_running())):
- index = self.dialog.table.model().index(row, 0)
- self.dialog.table.setCurrentIndex(index)
- return vm
- return None
- def _select_templatevm(self, running=None):
- for row in range(self.dialog.table.model().rowCount()):
- template = self._get_table_item(row, "Template")
- vm = self._get_table_vm(row)
- if template == 'TemplateVM' and \
- (running is None
- or (running and vm.is_running())
- or (not running and not vm.is_running())):
- index = self.dialog.table.model().index(row, 0)
- self.dialog.table.setCurrentIndex(index)
- return vm
- return None
- def __check_sorting(self, column_name):
- last_text = None
- last_vm = None
- for row in range(self.dialog.table.model().rowCount()):
- vm = self._get_table_item(row, "Name")
- text = self._get_table_item(row, column_name)
- if row == 0:
- self.assertEqual(vm, "dom0", "dom0 is not sorted first")
- elif last_text is None:
- last_text = text
- last_vm = vm
- else:
- if last_text == text:
- self.assertGreater(
- vm.lower(), last_vm.lower(),
- "Incorrect sorting for {}".format(column_name))
- else:
- self.assertGreater(
- text.lower(), last_text.lower(),
- "Incorrect sorting for {}".format(column_name))
- last_text = text
- last_vm = vm
- def _get_table_vminfo(self, row):
- model = self.dialog.table.model()
- return model.index(row, 0).data(Qt.UserRole)
- def _get_table_vm(self, row):
- model = self.dialog.table.model()
- return model.index(row, 0).data(Qt.UserRole).vm
- def _get_table_item(self, row, column_name, role = Qt.DisplayRole):
- model = self.dialog.table.model()
- column = self.dialog.qubes_model.columns_indices.index(column_name)
- return model.index(row, column).data(role)
- class QubeManagerThreadTest(unittest.TestCase):
- def test_01_startvm_thread(self):
- vm = unittest.mock.Mock(spec=['start'])
- thread = qube_manager.StartVMThread(vm)
- thread.run()
- vm.start.assert_called_once_with()
- def test_02_startvm_thread_error(self):
- vm = unittest.mock.Mock(
- spec=['start'],
- **{'start.side_effect': exc.QubesException('Error')})
- thread = qube_manager.StartVMThread(vm)
- thread.run()
- self.assertIsNotNone(thread.msg)
- def test_10_run_command_thread(self):
- vm = unittest.mock.Mock(spec=['run'])
- thread = qube_manager.RunCommandThread(vm, "test_command")
- thread.run()
- vm.run.assert_called_once_with("test_command")
- def test_11_run_command_thread_error(self):
- vm = unittest.mock.Mock(spec=['run'],
- **{'run.side_effect': ChildProcessError})
- thread = qube_manager.RunCommandThread(vm, "test_command")
- thread.run()
- self.assertIsNotNone(thread.msg)
- @unittest.mock.patch('subprocess.check_call')
- def test_20_update_vm_thread_dom0(self, check_call):
- vm = unittest.mock.Mock(spec=['qid'])
- vm.qid = 0
- thread = qube_manager.UpdateVMThread(vm)
- thread.run()
- check_call.assert_called_once_with(
- ["/usr/bin/qubes-dom0-update", "--clean", "--gui"])
- @unittest.mock.patch('builtins.open')
- @unittest.mock.patch('subprocess.call')
- def test_21_update_vm_thread_running(self, mock_call, mock_open):
- vm = unittest.mock.Mock(
- spec=['qid', 'is_running', 'run_service_for_stdio', 'run_service'],
- **{'is_running.return_value': True})
- vm.qid = 1
- vm.run_service_for_stdio.return_value = (b'changed=no\n', None)
- thread = qube_manager.UpdateVMThread(vm)
- thread.run()
- mock_open.assert_called_with(
- '/usr/libexec/qubes-manager/dsa-4371-update', 'rb')
- vm.run_service_for_stdio.assert_called_once_with(
- "qubes.VMShell", user='root', input=unittest.mock.ANY)
- vm.run_service.assert_called_once_with(
- "qubes.InstallUpdatesGUI", user="root", wait=False)
- self.assertEqual(mock_call.call_count, 0)
- @unittest.mock.patch('builtins.open')
- @unittest.mock.patch('subprocess.call')
- def test_22_update_vm_thread_not_running(self, mock_call, mock_open):
- vm = unittest.mock.Mock(
- spec=['qid', 'is_running', 'run_service_for_stdio',
- 'run_service', 'start', 'name'],
- **{'is_running.return_value': False})
- vm.qid = 1
- vm.run_service_for_stdio.return_value = (b'changed=yes\n', None)
- thread = qube_manager.UpdateVMThread(vm)
- thread.run()
- mock_open.assert_called_with(
- '/usr/libexec/qubes-manager/dsa-4371-update', 'rb')
- vm.start.assert_called_once_with()
- vm.run_service_for_stdio.assert_called_once_with(
- "qubes.VMShell", user='root', input=unittest.mock.ANY)
- vm.run_service.assert_called_once_with(
- "qubes.InstallUpdatesGUI", user="root", wait=False)
- self.assertEqual(mock_call.call_count, 1)
- @unittest.mock.patch('builtins.open')
- @unittest.mock.patch('subprocess.check_call')
- def test_23_update_vm_thread_error(self, *_args):
- vm = unittest.mock.Mock(
- spec=['qid', 'is_running'],
- **{'is_running.side_effect': ChildProcessError})
- vm.qid = 1
- thread = qube_manager.UpdateVMThread(vm)
- thread.run()
- self.assertIsNotNone(thread.msg)
- class VMShutdownMonitorTest(unittest.TestCase):
- @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
- @unittest.mock.patch('PyQt5.QtCore.QTimer')
- def test_01_vm_shutdown_correct(self, mock_timer, mock_question):
- mock_vm = unittest.mock.Mock()
- mock_vm.is_running.return_value = False
- monitor = qube_manager.VmShutdownMonitor(mock_vm)
- monitor.restart_vm_if_needed = unittest.mock.Mock()
- monitor.check_if_vm_has_shutdown()
- self.assertEqual(mock_question.call_count, 0)
- self.assertEqual(mock_timer.call_count, 0)
- monitor.restart_vm_if_needed.assert_called_once_with()
- @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
- @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
- def test_02_vm_not_shutdown_wait(self, mock_timer, mock_question):
- mock_question().clickedButton.return_value = 1
- mock_question().addButton.return_value = 0
- mock_vm = unittest.mock.Mock()
- mock_vm.is_running.return_value = True
- mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
- monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=1)
- time.sleep(3)
- monitor.check_if_vm_has_shutdown()
- self.assertEqual(mock_timer.call_count, 1)
- @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
- @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
- def test_03_vm_kill(self, mock_timer, mock_question):
- mock_question().clickedButton.return_value = 1
- mock_question().addButton.return_value = 1
- mock_vm = unittest.mock.Mock()
- mock_vm.is_running.return_value = True
- mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
- monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=1)
- time.sleep(3)
- monitor.restart_vm_if_needed = unittest.mock.Mock()
- monitor.check_if_vm_has_shutdown()
- self.assertEqual(mock_timer.call_count, 0)
- mock_vm.kill.assert_called_once_with()
- monitor.restart_vm_if_needed.assert_called_once_with()
- @unittest.mock.patch('qubesmanager.qube_manager.QMessageBox')
- @unittest.mock.patch('PyQt5.QtCore.QTimer.singleShot')
- def test_04_check_later(self, mock_timer, mock_question):
- mock_vm = unittest.mock.Mock()
- mock_vm.is_running.return_value = True
- mock_vm.start_time = datetime.datetime.now().timestamp() - 3000
- monitor = qube_manager.VmShutdownMonitor(mock_vm, shutdown_time=3000)
- time.sleep(1)
- monitor.check_if_vm_has_shutdown()
- self.assertEqual(mock_question.call_count, 0)
- self.assertEqual(mock_timer.call_count, 1)
- if __name__ == "__main__":
- ha_syslog = logging.handlers.SysLogHandler('/dev/log')
- ha_syslog.setFormatter(
- logging.Formatter('%(name)s[%(process)d]: %(message)s'))
- logging.root.addHandler(ha_syslog)
- unittest.main()
|