From 1ba8d7971f57487170033926aeb5a705726142c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Thu, 28 Feb 2019 00:32:44 +0100 Subject: [PATCH] tests: unit tests for internal.SuspendPre and internal.SuspendPost API --- qubes/tests/__init__.py | 1 + qubes/tests/api_internal.py | 166 ++++++++++++++++++++++++++++++++++++ rpm_spec/core-dom0.spec.in | 1 + 3 files changed, 168 insertions(+) create mode 100644 qubes/tests/api_internal.py diff --git a/qubes/tests/__init__.py b/qubes/tests/__init__.py index 0cda6b1b..31806bb6 100644 --- a/qubes/tests/__init__.py +++ b/qubes/tests/__init__.py @@ -1341,6 +1341,7 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument 'qubes.tests.api', 'qubes.tests.api_admin', 'qubes.tests.api_misc', + 'qubes.tests.api_internal', 'qubespolicy.tests', 'qubespolicy.tests.cli', ): diff --git a/qubes/tests/api_internal.py b/qubes/tests/api_internal.py new file mode 100644 index 00000000..cfdab4b4 --- /dev/null +++ b/qubes/tests/api_internal.py @@ -0,0 +1,166 @@ +# -*- encoding: utf-8 -*- +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2019 Marek Marczykowski-Górecki +# +# +# 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, see . +import asyncio +import subprocess +import qubes.api.internal +import qubes.tests +import qubes.vm.adminvm +from unittest import mock +from mock import call + +def mock_coro(f): + @asyncio.coroutine + def coro_f(*args, **kwargs): + return f(*args, **kwargs) + + return coro_f + +class TC_00_API_Misc(qubes.tests.QubesTestCase): + def setUp(self): + super(TC_00_API_Misc, self).setUp() + self.tpl = mock.NonCallableMagicMock(name='template') + del self.tpl.template + self.src = mock.NonCallableMagicMock(name='appvm', + template=self.tpl) + self.app = mock.NonCallableMock() + self.dest = mock.NonCallableMock() + self.dest.name = 'dom0' + self.app.configure_mock(domains={ + 'dom0': self.dest, + 'test-vm': self.src, + }) + + def configure_qdb(self, entries): + self.src.configure_mock(**{ + 'untrusted_qdb.read.side_effect': ( + lambda path: entries.get(path, None)), + 'untrusted_qdb.list.side_effect': ( + lambda path: sorted(entries.keys())), + }) + + def create_mockvm(self, features=None): + if features is None: + features = {} + vm = mock.Mock() + vm.features.check_with_template.side_effect = features.get + vm.run_service.return_value.wait = mock_coro( + vm.run_service.return_value.wait) + vm.run_service = mock_coro(vm.run_service) + vm.suspend = mock_coro(vm.suspend) + vm.resume = mock_coro(vm.resume) + return vm + + def call_mgmt_func(self, method, arg=b'', payload=b''): + mgmt_obj = qubes.api.internal.QubesInternalAPI(self.app, + b'dom0', method, b'dom0', arg) + + loop = asyncio.get_event_loop() + response = loop.run_until_complete( + mgmt_obj.execute(untrusted_payload=payload)) + return response + + def test_000_suspend_pre(self): + dom0 = mock.NonCallableMock(spec=qubes.vm.adminvm.AdminVM) + + running_vm = self.create_mockvm(features={'qrexec': True}) + running_vm.is_running.return_value = True + + not_running_vm = self.create_mockvm(features={'qrexec': True}) + not_running_vm.is_running.return_value = False + + no_qrexec_vm = self.create_mockvm() + no_qrexec_vm.is_running.return_value = True + + domains_dict = { + 'dom0': dom0, + 'running': running_vm, + 'not-running': not_running_vm, + 'no-qrexec': no_qrexec_vm, + } + self.addCleanup(domains_dict.clear) + self.app.domains = mock.MagicMock(**{ + '__iter__': lambda _: iter(domains_dict.values()), + '__getitem__': domains_dict.get, + }) + + ret = self.call_mgmt_func(b'internal.SuspendPre') + self.assertIsNone(ret) + self.assertFalse(dom0.called) + + self.assertNotIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY), + not_running_vm.mock_calls) + self.assertNotIn(('suspend', (), {}), + not_running_vm.mock_calls) + + self.assertIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY), + running_vm.mock_calls) + self.assertIn(('suspend', (), {}), + running_vm.mock_calls) + + self.assertNotIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY), + no_qrexec_vm.mock_calls) + self.assertIn(('suspend', (), {}), + no_qrexec_vm.mock_calls) + + def test_001_suspend_post(self): + dom0 = mock.NonCallableMock(spec=qubes.vm.adminvm.AdminVM) + + running_vm = self.create_mockvm(features={'qrexec': True}) + running_vm.is_running.return_value = True + running_vm.get_power_state.return_value = 'Suspended' + + not_running_vm = self.create_mockvm(features={'qrexec': True}) + not_running_vm.is_running.return_value = False + not_running_vm.get_power_state.return_value = 'Halted' + + no_qrexec_vm = self.create_mockvm() + no_qrexec_vm.is_running.return_value = True + no_qrexec_vm.get_power_state.return_value = 'Suspended' + + domains_dict = { + 'dom0': dom0, + 'running': running_vm, + 'not-running': not_running_vm, + 'no-qrexec': no_qrexec_vm, + } + self.addCleanup(domains_dict.clear) + self.app.domains = mock.MagicMock(**{ + '__iter__': lambda _: iter(domains_dict.values()), + '__getitem__': domains_dict.get, + }) + + ret = self.call_mgmt_func(b'internal.SuspendPost') + self.assertIsNone(ret) + self.assertFalse(dom0.called) + + self.assertNotIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY), + not_running_vm.mock_calls) + self.assertNotIn(('resume', (), {}), + not_running_vm.mock_calls) + + self.assertIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY), + running_vm.mock_calls) + self.assertIn(('resume', (), {}), + running_vm.mock_calls) + + self.assertNotIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY), + no_qrexec_vm.mock_calls) + self.assertIn(('resume', (), {}), + no_qrexec_vm.mock_calls) diff --git a/rpm_spec/core-dom0.spec.in b/rpm_spec/core-dom0.spec.in index 1e4c5987..6c8f6e69 100644 --- a/rpm_spec/core-dom0.spec.in +++ b/rpm_spec/core-dom0.spec.in @@ -292,6 +292,7 @@ fi %{python3_sitelib}/qubes/tests/api.py %{python3_sitelib}/qubes/tests/api_admin.py +%{python3_sitelib}/qubes/tests/api_internal.py %{python3_sitelib}/qubes/tests/api_misc.py %{python3_sitelib}/qubes/tests/app.py %{python3_sitelib}/qubes/tests/devices.py