manager/qubesmanager/tests/test_backup.py

#!/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 logging.handlers
import unittest
import unittest.mock

from PyQt5 import QtTest, QtCore, QtWidgets
from qubesadmin import Qubes, events, utils, exc
from qubesmanager import backup
from qubesmanager.tests import init_qtapp


class BackupTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        qapp = Qubes()

        cls.dom0_name = "dom0"

        cls.vms = []
        cls.running_vm = None

        for vm in qapp.domains:
            if vm.klass != "AdminVM" and vm.is_running():
                cls.running_vm = vm.name
            if vm.klass != "AdminVM" and vm.get_disk_utilization() > 0:
                cls.vms.append(vm.name)
            if cls.running_vm and len(cls.vms) >= 3:
                break

    def setUp(self):
        super(BackupTest, self).setUp()
        self.qtapp, self.loop = init_qtapp()

        # mock up nonexistence of saved backup settings
        self.patcher_open = unittest.mock.patch('builtins.open')
        self.mock_open = self.patcher_open.start()
        self.mock_open.side_effect = FileNotFoundError()
        self.addCleanup(self.patcher_open.stop)

        # mock up the Backup Thread to avoid accidentally changing system state
        self.patcher_thread = unittest.mock.patch(
            'qubesmanager.backup.BackupThread')
        self.mock_thread = self.patcher_thread.start()
        self.addCleanup(self.patcher_thread.stop)

        self.qapp = Qubes()
        self.dispatcher = events.EventsDispatcher(self.qapp)

        self.dialog = backup.BackupVMsWindow(
            self.qtapp, self.qapp, self.dispatcher)
        self.dialog.show()

    def tearDown(self):
        self.dialog.done(0)
        super(BackupTest, self).tearDown()

    def test_00_window_loads(self):
        self.assertTrue(self.dialog.select_vms_widget is not None)

    def test_01_vms_load_correctly(self):
        all_vms = len([vm for vm in self.qapp.domains
                       if not vm.features.get('internal', False)])

        selected_vms = self.dialog.select_vms_widget.selected_list.count()
        available_vms = self.dialog.select_vms_widget.available_list.count()

        self.assertEqual(all_vms, available_vms + selected_vms)

    def test_02_correct_defaults(self):
        # backup is compressed
        self.assertTrue(self.dialog.compress_checkbox.isChecked(),
                        "Compress backup should be checked by default")

        # correct VMs are selected
        include_in_backups_no = len(
            [vm for vm in self.qapp.domains
             if not vm.features.get('internal', False)
             and getattr(vm, 'include_in_backups', True)])
        selected_no = self.dialog.select_vms_widget.selected_list.count()
        self.assertEqual(include_in_backups_no, selected_no,
                         "Incorrect VMs selected by default")

        # passphrase is empty
        self.assertEqual(self.dialog.passphrase_line_edit.text(), "",
                         "Passphrase should be empty")

        # save defaults
        self.assertTrue(self.dialog.save_profile_checkbox.isChecked(),
                        "By default, profile should be saved")

    def test_03_select_vms_widget(self):
        number_of_all_vms = len([vm for vm in self.qapp.domains
                                 if not vm.features.get('internal', False)])

        # select all
        self.dialog.select_vms_widget.add_all_button.click()
        self.assertEqual(number_of_all_vms,
                         self.dialog.select_vms_widget.selected_list.count(),
                         "Add All VMs does not work")

        # remove all
        self.dialog.select_vms_widget.remove_all_button.click()
        self.assertEqual(number_of_all_vms,
                         self.dialog.select_vms_widget.available_list.count(),
                         "Remove All VMs does not work")

        self._select_vm(self.vms[0])

        self.assertEqual(self.dialog.select_vms_widget.selected_list.count(),
                         1, "Select a single VM does not work")

    def test_04_open_directory(self):
        self.dialog.next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        with unittest.mock.patch('qubesmanager.backup_utils.'
                                 'select_path_button_clicked') as mock_func:
            self.dialog.select_path_button.click()
            mock_func.assert_called_once_with(unittest.mock.ANY)

    def test_05_running_vms_listed(self):
        self.dialog.next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        running_vms = [vm.name for vm in self.qapp.domains if vm.is_running()]

        listed_vms = []

        for i in range(self.dialog.appvm_combobox.count()):
            listed_vms.append(self.dialog.appvm_combobox.itemText(i))

        self.assertListEqual(sorted(running_vms), sorted(listed_vms),
                             "Incorrect list of running vms")

    def test_06_passphrase_verification(self):
        self.dialog.next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)
        # required to check if next button is correctly enabled
        self.dialog.dir_line_edit.setText("/home")

        next_button = self.dialog.button(self.dialog.NextButton)

        # check if next remains inactive for various incorrect
        # passphrase/incorrect combinations
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("fail")
        self.assertFalse(next_button.isEnabled(),
                         "Mismatched passphrase/verification accepted")

        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("")
        self.assertFalse(next_button.isEnabled(), "Empty verification accepted")

        self.dialog.passphrase_line_edit.setText("")
        self.dialog.passphrase_line_edit_verify.setText("fail")
        self.assertFalse(next_button.isEnabled(), "Empty passphrase accepted")

        self.dialog.passphrase_line_edit.setText("")
        self.dialog.passphrase_line_edit_verify.setText("")
        self.assertFalse(next_button.isEnabled(),
                         "Empty passphrase and verification accepted")

        # check if next is active for a correct passphrase/verify
        # combination
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.assertTrue(next_button.isEnabled(),
                        "Matching passphrase/verification not accepted")

    def test_07_disk_space_correct(self):
        for i in range(self.dialog.select_vms_widget.available_list.count()):
            item = self.dialog.select_vms_widget.available_list.item(i)
            if item.vm.name == self.dom0_name or \
                    item.vm.get_disk_utilization() > 0:
                self.assertGreater(
                    item.size, 0,
                    "{} size incorrectly reported as 0".format(item.vm.name))

    def test_08_total_size_correct(self):
        if len(self.vms) < 3:
            self.skipTest("Insufficient number of VMs with positive "
                          "disk utilization")
        # select nothing
        self.dialog.select_vms_widget.remove_all_button.click()
        self.assertEqual(self.dialog.total_size_label.text(), "0",
                         "Total size of 0 vms incorrectly reported as 0")

        current_size = 0
        # select a single VM
        self._select_vm(self.vms[0])

        current_size += self.qapp.domains[self.vms[0]].get_disk_utilization()
        self.assertEqual(self.dialog.total_size_label.text(),
                         utils.size_to_human(current_size),
                         "Size incorrectly listed for a single VM")

        # add two more
        self._select_vm(self.vms[1])
        self._select_vm(self.vms[2])

        current_size += self.qapp.domains[self.vms[1]].get_disk_utilization()
        current_size += self.qapp.domains[self.vms[2]].get_disk_utilization()

        self.assertEqual(self.dialog.total_size_label.text(),
                         utils.size_to_human(current_size),
                         "Size incorrectly listed for several VMs")

        # remove one
        self._deselect_vm(self.vms[0])
        current_size -= self.qapp.domains[self.vms[0]].get_disk_utilization()
        self.assertEqual(self.dialog.total_size_label.text(),
                         utils.size_to_human(current_size),
                         "Size incorrectly listed for several VMs")

    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_10_first_backup(self, mock_qubesd, mock_write_profile):
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_vms_page)

        self.dialog.select_vms_widget.remove_all_button.click()
        self._select_vm(self.vms[0])

        self._click_next()

        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        # setup backup
        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.dialog.save_profile_checkbox.setChecked(True)
        self.dialog.turn_off_checkbox.setChecked(False)
        self.dialog.compress_checkbox.setChecked(False)
        expected_settings = {'destination_vm': self.dom0_name,
                             'destination_path': "/home",
                             'include': [self.vms[0]],
                             'passphrase_text': "pass",
                             'compression': False}
        with unittest.mock.patch.object(self.dialog.textEdit, 'setText')\
                as mock_set_text:
            self._click_next()

            # make sure the confirmation is not empty
            self.assertTrue(self.dialog.currentPage()
                            is self.dialog.confirm_page)

            mock_write_profile.assert_called_with(expected_settings, True)
            mock_qubesd.assert_called_with('dom0', 'admin.backup.Info',
                                           unittest.mock.ANY)
            mock_set_text.assert_called_once_with("backup output")

        # make sure the backup is executed
        self._click_next()
        self.mock_thread.assert_called_once_with(
            self.qapp.domains[self.dom0_name])
        self.mock_thread().start.assert_called_once_with()

    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_11_second_backup(self, mock_qubesd, mock_write_profile):
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_vms_page)

        self.dialog.select_vms_widget.remove_all_button.click()
        self._select_vm(self.dom0_name)
        self._select_vm(self.vms[0])
        self._select_vm(self.vms[1])

        self._click_next()

        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        # setup backup
        self._select_location(self.running_vm)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("longerPassPhrase")
        self.dialog.passphrase_line_edit_verify.setText("longerPassPhrase")
        self.dialog.save_profile_checkbox.setChecked(False)
        self.dialog.turn_off_checkbox.setChecked(False)
        self.dialog.compress_checkbox.setChecked(True)
        expected_settings = {'destination_vm': self.running_vm,
                             'destination_path': "/home",
                             'include': sorted([self.dom0_name, self.vms[0],
                                         self.vms[1]]),
                             'passphrase_text': "longerPassPhrase",
                             'compression': True}
        with unittest.mock.patch.object(self.dialog.textEdit, 'setText')\
                as mock_set_text:
            self._click_next()

            # make sure the confirmation is not empty
            self.assertTrue(self.dialog.currentPage()
                            is self.dialog.confirm_page)

            mock_write_profile.assert_called_with(expected_settings, True)
            mock_qubesd.assert_called_with('dom0', 'admin.backup.Info',
                                           unittest.mock.ANY)
            mock_set_text.assert_called_once_with("backup output")

        # make sure the backup is executed
        self._click_next()
        self.mock_thread.assert_called_once_with(
            self.qapp.domains[self.running_vm])
        self.mock_thread().start.assert_called_once_with()

    @unittest.mock.patch('qubesmanager.backup_utils.load_backup_profile')
    def test_20_loading_settings(self, mock_load):

        mock_load.return_value = {
            'destination_vm': self.running_vm,
            'destination_path': "/home",
            'include': [self.dom0_name, self.vms[0], self.vms[1]],
            'passphrase_text': "longerPassPhrase",
            'compression': True
        }

        self.dialog.hide()
        self.dialog.deleteLater()
        self.qtapp.processEvents()

        self.dialog = backup.BackupVMsWindow(
            self.qtapp, self.qapp, self.dispatcher)
        self.dialog.show()

        # check if settings were loaded
        self.assertEqual(self.dialog.appvm_combobox.currentText(),
                         self.running_vm,
                         "Destination VM not loaded")
        self.assertEqual(self.dialog.dir_line_edit.text(), "/home",
                         "Destination path not loaded")
        self.assertEqual(self.dialog.passphrase_line_edit.text(),
                         "longerPassPhrase", "Passphrase not loaded")
        self.assertEqual(self.dialog.passphrase_line_edit_verify.text(),
                         "longerPassPhrase", "Passphrase verify not loaded")
        self.assertTrue(self.dialog.compress_checkbox.isChecked())

        # check that 'include' vms were not pre-selected
        include_in_backups_no = len(
            [vm for vm in self.qapp.domains
             if not vm.features.get('internal', False)
             and getattr(vm, 'include_in_backups', True)])
        selected_no = self.dialog.select_vms_widget.selected_list.count()
        self.assertEqual(include_in_backups_no, selected_no,
                         "Incorrect VM list selected")

        # check no errors were detected
        self.assertFalse(self.dialog.unrecognized_config_label.isVisible())

    @unittest.mock.patch('qubesmanager.backup_utils.load_backup_profile')
    def test_21_loading_settings_error(self, mock_load):

        mock_load.return_value = {
            'destination_vm': "incorrect_vm",
        }

        self.dialog.hide()
        self.dialog.deleteLater()
        self.qtapp.processEvents()

        self.dialog = backup.BackupVMsWindow(
            self.qtapp, self.qapp, self.dispatcher)
        self.dialog.show()

        # check errors were detected
        self.assertTrue(self.dialog.unrecognized_config_label.isVisible())

    @unittest.mock.patch('qubesmanager.backup_utils.load_backup_profile')
    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.information')
    def test_22_loading_settings_exc(self, mock_info, mock_load):

        mock_load.side_effect = exc.QubesException('Error')

        self.dialog.hide()
        self.dialog.deleteLater()
        self.qtapp.processEvents()

        self.dialog = backup.BackupVMsWindow(
            self.qtapp, self.qapp, self.dispatcher)
        self.dialog.show()

        # check error was reported
        self.assertEqual(mock_info.call_count, 1, "Warning not shown")

    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_23_cancel_confirm(self, *_args):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")

        self._click_next()

        # attempt to cancel
        with unittest.mock.patch('os.remove') as mock_remove:
            self._click_cancel()
            mock_remove.assert_called_once_with(
                '/etc/qubes/backup/qubes-manager-backup-tmp.conf')

    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_24_cancel_in_progress(self, mock_call, *_args):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")

        self._click_next()
        self._click_next()

        # attempt to cancel
        with unittest.mock.patch('os.remove') as mock_remove:
            self._click_cancel()
            mock_call.assert_called_with('dom0', 'admin.backup.Cancel',
                                         'qubes-manager-backup-tmp')
            mock_remove.assert_called_once_with(
                '/etc/qubes/backup/qubes-manager-backup-tmp.conf')

    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
    @unittest.mock.patch('os.system')
    @unittest.mock.patch('os.remove')
    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_25_successful_backup(self, _a, _b, mock_remove,
                                  mock_system, mock_warning):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.dialog.turn_off_checkbox.setChecked(False)

        self._click_next()
        self._click_next()

        # assume backup went correctly
        self.mock_thread().msg = None

        self.mock_thread().finished.connect.assert_called_once_with(
            self.dialog.backup_finished)

        self.dialog.backup_finished()

        self.assertFalse(self.dialog.button(
            self.dialog.CancelButton).isEnabled())
        self.assertTrue(self.dialog.button(
            self.dialog.FinishButton).isEnabled())
        mock_remove.assert_called_once_with(
            '/etc/qubes/backup/qubes-manager-backup-tmp.conf')
        self.assertEqual(mock_system.call_count, 0,
                         "System turned off unnecessarily")
        self.assertEqual(mock_warning.call_count, 0,
                         "Backup succeeded but received warning")

    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
    @unittest.mock.patch('os.system')
    @unittest.mock.patch('os.remove')
    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_26_success_backup_poweroff(
            self, _a, _b, mock_remove, mock_system, mock_warning):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.dialog.turn_off_checkbox.setChecked(True)

        self._click_next()

        self._click_next()

        # assume backup went correctly
        self.mock_thread().msg = None
        self.mock_thread().finished.connect.assert_called_once_with(
            self.dialog.backup_finished)

        self.dialog.backup_finished()

        self.assertFalse(self.dialog.button(
            self.dialog.CancelButton).isEnabled())
        self.assertTrue(self.dialog.button(
            self.dialog.FinishButton).isEnabled())
        mock_remove.assert_called_once_with(
            '/etc/qubes/backup/qubes-manager-backup-tmp.conf')
        mock_system.assert_called_once_with('systemctl poweroff')
        self.assertEqual(mock_warning.call_count, 0,
                         "Backup succeeded but received warning")

    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
    @unittest.mock.patch('os.system')
    @unittest.mock.patch('os.remove')
    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_27_failed_backup(
            self, _a, _b, mock_remove, mock_system, mock_warn):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.dialog.turn_off_checkbox.setChecked(True)

        self._click_next()
        self._click_next()

        # assume backup went wrong
        self.mock_thread().msg = "Error"
        self.mock_thread().finished.connect.assert_called_once_with(
            self.dialog.backup_finished)

        self.dialog.backup_finished()

        self.assertFalse(self.dialog.button(
            self.dialog.CancelButton).isEnabled())
        self.assertTrue(self.dialog.button(
            self.dialog.FinishButton).isEnabled())
        mock_remove.assert_called_once_with(
            '/etc/qubes/backup/qubes-manager-backup-tmp.conf')
        self.assertEqual(mock_system.call_count, 0,
                         "Attempted shutdown at failed backup")
        self.assertEqual(mock_warn.call_count, 1)

    @unittest.mock.patch('PyQt5.QtWidgets.QMessageBox.warning')
    @unittest.mock.patch('os.system')
    @unittest.mock.patch('os.remove')
    @unittest.mock.patch('qubesmanager.backup_utils.write_backup_profile')
    @unittest.mock.patch('qubesadmin.Qubes.qubesd_call',
                         return_value=b'backup output')
    def test_28_progress(
            self, _a, _b, _mock_remove, _mock_system, _mock_warn):
        self._click_next()
        self.assertTrue(self.dialog.currentPage()
                        is self.dialog.select_dir_page)

        self._select_location(self.dom0_name)
        self.dialog.dir_line_edit.setText("/home")
        self.dialog.passphrase_line_edit.setText("pass")
        self.dialog.passphrase_line_edit_verify.setText("pass")
        self.dialog.turn_off_checkbox.setChecked(True)

        self._click_next()
        self._click_next()

        # see if backup is correctly in progress
        self.assertTrue(self.dialog.button(
            self.dialog.CancelButton).isEnabled())
        self.assertFalse(self.dialog.button(
            self.dialog.FinishButton).isEnabled())
        self.assertEqual(self.dialog.progress_bar.value(), 0,
                         "Progress bar does not start at 0")

        # this is not a perfect method, but it is something

        self.dialog.on_backup_progress(None, None, progress='23.3123')
        self.assertEqual(self.dialog.progress_bar.value(), 23,
                         "Progress bar does not update correctly")

        self.dialog.on_backup_progress(None, None, progress='87.89')
        self.assertEqual(self.dialog.progress_bar.value(), 87,
                         "Progress bar does not update correctly")

    def _select_location(self, vm_name):
        widget = self.dialog.appvm_combobox
        widget.setCurrentIndex(0)
        while not widget.currentText() == vm_name:
            if widget.currentIndex() == widget.count():
                self.skipTest("target VM not found")
            widget.setCurrentIndex(widget.currentIndex() + 1)

    def _click_next(self):
        next_widget = self.dialog.button(QtWidgets.QWizard.NextButton)
        QtTest.QTest.mouseClick(next_widget, QtCore.Qt.LeftButton)

    def _click_cancel(self):
        cancel_widget = self.dialog.button(QtWidgets.QWizard.CancelButton)
        QtTest.QTest.mouseClick(cancel_widget, QtCore.Qt.LeftButton)

    def _select_vm(self, name_starts_with):
        for i in range(self.dialog.select_vms_widget.available_list.count()):
            item = self.dialog.select_vms_widget.available_list.item(i)
            if item.text().startswith(name_starts_with):
                item.setSelected(True)
                self.dialog.select_vms_widget.add_selected_button.click()
                return

    def _deselect_vm(self, name_starts_with):
        for i in range(self.dialog.select_vms_widget.selected_list.count()):
            item = self.dialog.select_vms_widget.selected_list.item(i)
            if item.text().startswith(name_starts_with):
                item.setSelected(True)
                self.dialog.select_vms_widget.remove_selected_button.click()
                return


class BackupThreadTest(unittest.TestCase):
    def test_01_backup_thread_vm_on(self):
        vm = unittest.mock.Mock(spec=['is_running', 'app'],
                                **{'is_running.return_value': True})

        vm.app = unittest.mock.Mock()

        thread = backup.BackupThread(vm)
        thread.run()

        vm.app.qubesd_call.assert_called_with(
            'dom0', 'admin.backup.Execute', 'qubes-manager-backup-tmp')

    def test_02_backup_thread_vm_off(self):
        vm = unittest.mock.Mock(spec=['is_running', 'app', 'start'],
                                **{'is_running.return_value': False})

        vm.app = unittest.mock.Mock()

        thread = backup.BackupThread(vm)
        thread.run()

        vm.app.qubesd_call.assert_called_with(
            'dom0', 'admin.backup.Execute', 'qubes-manager-backup-tmp')
        vm.start.assert_called_once_with()

    def test_03_backup_thread_error(self):
        vm = unittest.mock.Mock(spec=['is_running', 'app'],
                                **{'is_running.return_value': True})

        vm.app = unittest.mock.Mock()
        vm.app.qubesd_call.side_effect = exc.QubesException('Error')

        thread = backup.BackupThread(vm)
        thread.run()

        self.assertIsNotNone(thread.msg)


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()