8eb61b11ae
Settings unavailable due to permissions will be unavailable in the tool, but the tool will start and attempt to show as much as possible.
1466 lines
58 KiB
Python
1466 lines
58 KiB
Python
#!/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.clone_vm.CloneVMDlg')
|
|
def test_232_clonevm(self, mock_clone):
|
|
action = self.dialog.action_clonevm
|
|
|
|
self._select_admin_vm()
|
|
self.assertFalse(action.isEnabled())
|
|
|
|
selected_vm = self._select_non_admin_vm()
|
|
self.assertTrue(action.isEnabled())
|
|
|
|
action.trigger()
|
|
mock_clone.assert_called_once_with(self.qtapp, self.qapp,
|
|
src_vm=selected_vm)
|
|
|
|
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=['klass'])
|
|
vm.klass = 'AdminVM'
|
|
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=['klass', 'is_running', 'run_service_for_stdio', 'run_service'],
|
|
**{'is_running.return_value': True})
|
|
|
|
vm.klass = 'AppVM'
|
|
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=['klass', 'is_running', 'run_service_for_stdio',
|
|
'run_service', 'start', 'name'],
|
|
**{'is_running.return_value': False})
|
|
|
|
vm.klass = 'AppVM'
|
|
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=['klass', 'is_running'],
|
|
**{'is_running.side_effect': ChildProcessError})
|
|
|
|
vm.klass = 'AppVM'
|
|
|
|
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()
|