From ed93b40e726faac892a465b8220ab033d3e9ceda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Fri, 21 Apr 2017 03:18:07 +0200 Subject: [PATCH] tools: qvm-check The tool and tests. --- qubesmgmt/tests/tools/qvm_check.py | 228 +++++++++++++++++++++++++++++ qubesmgmt/tools/qvm_check.py | 79 ++++++++++ 2 files changed, 307 insertions(+) create mode 100644 qubesmgmt/tests/tools/qvm_check.py create mode 100644 qubesmgmt/tools/qvm_check.py diff --git a/qubesmgmt/tests/tools/qvm_check.py b/qubesmgmt/tests/tools/qvm_check.py new file mode 100644 index 0000000..1398f1f --- /dev/null +++ b/qubesmgmt/tests/tools/qvm_check.py @@ -0,0 +1,228 @@ +# -*- encoding: utf8 -*- +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2017 Marek Marczykowski-Górecki +# +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License along +# with this program; if not, see . + +import qubesmgmt.tests +import qubesmgmt.tools.qvm_check + +class TC_00_qvm_check(qubesmgmt.tests.QubesTestCase): + def test_000_exists(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['some-vm'], app=self.app), + 0) + self.assertAllCalled() + + def test_001_exists_multi(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'other-vm class=AppVM state=Running\n' + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['some-vm', 'other-vm'], + app=self.app), + 0) + self.assertAllCalled() + + def test_002_exists_verbose(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['some-vm'], app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm exists\n') + self.assertAllCalled() + + def test_003_exists_multi_verbose(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'other-vm class=AppVM state=Running\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['some-vm', 'other-vm'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VMs other-vm, some-vm exist\n') + self.assertAllCalled() + + def test_004_running_verbose(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Running\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--running', + 'some-vm'], app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm is running\n') + self.assertAllCalled() + + def test_005_running_multi_verbose(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Running\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + self.app.expected_calls[ + ('some-vm2', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm2 class=AppVM state=Running\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--running', + 'some-vm', 'some-vm2'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VMs some-vm, some-vm2 are running\n') + self.assertAllCalled() + + def test_006_running_multi_verbose2(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Running\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + self.app.expected_calls[ + ('some-vm2', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm2 class=AppVM state=Running\n' + self.app.expected_calls[ + ('some-vm3', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm3 class=AppVM state=Halted\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--running', + '--all'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VMs some-vm, some-vm2 are running\n') + self.assertAllCalled() + + def test_007_not_running_verbose(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Running\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm3', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm3 class=AppVM state=Halted\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--running', + 'some-vm3'], + app=self.app), + 1) + self.assertEqual(stdout.getvalue(), + 'None of given VM is running\n') + self.assertAllCalled() + + def test_008_paused(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Paused\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm2', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm2 class=AppVM state=Paused\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--paused', + 'some-vm2'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm2 is paused\n') + self.assertAllCalled() + + def test_009_paused_multi(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Paused\n' \ + b'some-vm3 class=AppVM state=Halted\n' + self.app.expected_calls[ + ('some-vm2', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm2 class=AppVM state=Paused\n' + self.app.expected_calls[ + ('some-vm', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--paused', + 'some-vm2', 'some-vm'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm2 is paused\n') + self.assertAllCalled() + + + def test_010_template(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Paused\n' \ + b'some-vm3 class=TemplateVM state=Halted\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--template', + 'some-vm3'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm3 is a template\n') + self.assertAllCalled() + + def test_011_template_multi(self): + self.app.expected_calls[ + ('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00some-vm class=AppVM state=Running\n' \ + b'some-vm2 class=AppVM state=Paused\n' \ + b'some-vm3 class=TemplateVM state=Halted\n' + with qubesmgmt.tests.tools.StdoutBuffer() as stdout: + self.assertEqual( + qubesmgmt.tools.qvm_check.main(['--template', + 'some-vm2', 'some-vm3'], + app=self.app), + 0) + self.assertEqual(stdout.getvalue(), + 'VM some-vm3 is a template\n') + self.assertAllCalled() + diff --git a/qubesmgmt/tools/qvm_check.py b/qubesmgmt/tools/qvm_check.py new file mode 100644 index 0000000..1f292b0 --- /dev/null +++ b/qubesmgmt/tools/qvm_check.py @@ -0,0 +1,79 @@ +# pylint: disable=too-few-public-methods + +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser 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. +# +''' Exits sucessfull if the provided domains exists, else returns failure ''' + +from __future__ import print_function + +import sys + +import qubesmgmt.tools +import qubesmgmt.vm + +parser = qubesmgmt.tools.QubesArgumentParser(description=__doc__, + vmname_nargs='+') +parser.add_argument("--running", action="store_true", dest="running", + default=False, help="Determine if (any of given) VM is running") +parser.add_argument("--paused", action="store_true", dest="paused", + default=False, help="Determine if (any of given) VM is paused") +parser.add_argument("--template", action="store_true", dest="template", + default=False, help="Determine if (any of given) VM is a template") + + +def print_msg(domains, what_single, what_plural): + '''Print message in appropriate form about given domain(s)''' + # pylint: disable=len-as-condition + if len(domains) == 0: + print("None of given VM {!s}".format(what_single)) + elif len(domains) == 1: + print("VM {!s} {!s}".format(domains[0], what_single)) + else: + txt = ", ".join([vm.name for vm in sorted(domains)]) + print("VMs {!s} {!s}".format(txt, what_plural)) + + +def main(args=None, app=None): + '''Main function of qvm-check tool''' + args = parser.parse_args(args, app=app) + domains = args.domains + if args.running: + running = [vm for vm in domains if vm.is_running()] + if args.verbose: + print_msg(running, "is running", "are running") + return 0 if running else 1 + elif args.paused: + paused = [vm for vm in domains if vm.is_paused()] + if args.verbose: + print_msg(paused, "is paused", "are paused") + return 0 if paused else 1 + elif args.template: + template = [vm for vm in domains if isinstance(vm, + qubesmgmt.vm.TemplateVM)] + if args.verbose: + print_msg(template, "is a template", "are templates") + return 0 if template else 1 + else: + if args.verbose: + print_msg(domains, "exists", "exist") + return 0 if domains else 1 + +if __name__ == '__main__': + sys.exit(main())