avocado-framework/avocado

View on GitHub
selftests/unit/utils/test_service.py

Summary

Maintainability
D
1 day
Test Coverage
#  Copyright(c) 2013 Intel Corporation.
#
#  This program is free software; you can redistribute it and/or modify it
#  under the terms and conditions of the GNU General Public License,
#  version 2, as published by the Free Software Foundation.
#
#  This program is distributed in the hope 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 St - Fifth Floor, Boston, MA 02110-1301 USA.
#
#  The full GNU General Public License is included in this distribution in
#  the file called "COPYING".

import unittest.mock

from avocado.utils import service
from selftests.utils import setup_avocado_loggers

setup_avocado_loggers()


class TestMultipleInstances(unittest.TestCase):

    def test_different_runners(self):
        # Call 'set_target' on first runner
        runner1 = unittest.mock.Mock()
        runner1.return_value.stdout = 'systemd'
        service1 = service.service_manager(run=runner1)
        service1.set_target('foo_target')
        self.assertEqual(runner1.call_args[0][0],  # pylint: disable=E1136
                         'systemctl isolate foo_target')
        # Call 'start' on second runner
        runner2 = unittest.mock.Mock()
        runner2.return_value.stdout = 'init'
        service2 = service.service_manager(run=runner2)
        service2.start('foo_service')
        self.assertEqual(runner2.call_args[0][0],  # pylint: disable=E1136
                         'service foo_service start')


class TestSystemd(unittest.TestCase):

    def setUp(self):
        self.service_name = "fake_service"
        init_name = "systemd"
        command_generator = service._COMMAND_GENERATORS[init_name]
        self.service_command_generator = service._ServiceCommandGenerator(
            command_generator)

    def test_all_commands(self):
        # Test all commands except "set_target" which is tested elsewhere
        for cmd, _ in ((c, r) for (c, r) in
                       self.service_command_generator.commands if
                       c != "set_target"):
            ret = getattr(
                self.service_command_generator, cmd)(self.service_name)
            if cmd == "is_enabled":
                cmd = "is-enabled"
            if cmd == "reset_failed":
                cmd = "reset-failed"
            if cmd == "list":
                self.assertEqual(ret, ['systemctl', 'list-unit-files',
                                       '--type=service', '--no-pager',
                                       '--full'])
            else:
                self.assertEqual(ret, ["systemctl", cmd, "%s.service" %
                                       self.service_name])

    def test_set_target(self):
        ret = getattr(
            self.service_command_generator, "set_target")("multi-user.target")
        self.assertEqual(ret, ["systemctl", "isolate", "multi-user.target"])


class TestSysVInit(unittest.TestCase):

    def setUp(self):
        self.service_name = "fake_service"
        init_name = "init"
        command_generator = service._COMMAND_GENERATORS[init_name]
        self.service_command_generator = service._ServiceCommandGenerator(
            command_generator)

    def test_all_commands(self):
        command_name = "service"
        # Test all commands except "set_target" which is tested elsewhere
        for cmd, _ in ((c, r) for (c, r) in
                       self.service_command_generator.commands
                       if c != 'set_target'):
            ret = getattr(
                self.service_command_generator, cmd)(self.service_name)
            if cmd in ['set_target', 'reset_failed', 'mask', 'unmask']:
                exp = ['true']
            elif cmd == 'list':
                exp = ['chkconfig', '--list']
            else:
                if cmd == "is_enabled":
                    command_name = "chkconfig"
                    cmd = ""
                elif cmd == 'enable':
                    command_name = "chkconfig"
                    cmd = "on"
                elif cmd == 'disable':
                    command_name = "chkconfig"
                    cmd = "off"
                exp = [command_name, self.service_name, cmd]
            self.assertEqual(ret, exp)

    def test_set_target(self):
        ret = getattr(
            self.service_command_generator, "set_target")("multi-user.target")
        self.assertEqual(ret, ["telinit", "3"])


class TestSpecificServiceManager(unittest.TestCase):

    def setUp(self):
        self.run_mock = unittest.mock.Mock()
        self.init_name = "init"
        get_name_of_init_mock = unittest.mock.Mock(return_value="init")

        @unittest.mock.patch.object(service, "get_name_of_init",
                                    get_name_of_init_mock)
        def patch_service_command_generator():
            return service._auto_create_specific_service_command_generator()

        @unittest.mock.patch.object(service, "get_name_of_init",
                                    get_name_of_init_mock)
        def patch_service_result_parser():
            return service._auto_create_specific_service_result_parser()
        service_command_generator = patch_service_command_generator()
        service_result_parser = patch_service_result_parser()
        self.service_manager = service._SpecificServiceManager(
            "boot.lldpad", service_command_generator,
            service_result_parser, self.run_mock)

    def test_start(self):
        srv = "lldpad"
        self.service_manager.start()
        cmd = "service boot.%s start" % srv
        self.assertEqual(self.run_mock.call_args[0][0], cmd)  # pylint: disable=E1136

    def test_stop_with_args(self):
        srv = "lldpad"
        self.service_manager.stop(ignore_status=True)
        cmd = "service boot.%s stop" % srv
        self.assertEqual(self.run_mock.call_args[0][0], cmd)  # pylint: disable=E1136

    def test_list_is_not_present_in_SpecifcServiceManager(self):
        self.assertFalse(hasattr(self.service_manager, "list"))

    def test_set_target_is_not_present_in_SpecifcServiceManager(self):
        self.assertFalse(hasattr(self.service_manager, "set_target"))


def get_service_manager_from_init_and_run(init_name, run_mock):
    command_generator = service._COMMAND_GENERATORS[init_name]
    result_parser = service._RESULT_PARSERS[init_name]
    service_manager = service._SERVICE_MANAGERS[init_name]
    service_command_generator = service._ServiceCommandGenerator(
        command_generator)
    service_result_parser = service._ServiceResultParser(result_parser)
    return service_manager(service_command_generator, service_result_parser,
                           run_mock)


class TestSystemdServiceManager(unittest.TestCase):

    def setUp(self):
        self.run_mock = unittest.mock.Mock()
        self.init_name = "systemd"
        self.service_manager = get_service_manager_from_init_and_run(
            self.init_name,
            self.run_mock)

    def test_start(self):
        srv = "lldpad"
        self.service_manager.start(srv)
        cmd = ("systemctl start %s.service" % srv)
        self.assertEqual(self.run_mock.call_args[0][0], cmd)  # pylint: disable=E1136

    def test_list(self):
        list_result_mock = unittest.mock.Mock(
            exit_status=0,
            stdout_text="sshd.service enabled\n"
            "vsftpd.service disabled\n"
            "systemd-sysctl.service static\n")
        run_mock = unittest.mock.Mock(return_value=list_result_mock)
        service_manager = get_service_manager_from_init_and_run(self.init_name,
                                                                run_mock)
        list_result = service_manager.list(ignore_status=False)
        self.assertEqual(run_mock.call_args[0][0],  # pylint: disable=E1136
                         "systemctl list-unit-files --type=service "
                         "--no-pager --full")
        self.assertEqual(list_result, {'sshd': "enabled",
                                       'vsftpd': "disabled",
                                       'systemd-sysctl': "static"})

    def test_set_default_runlevel(self):
        runlevel = service.convert_sysv_runlevel(3)
        mktemp_mock = unittest.mock.Mock(return_value="temp_filename")
        symlink_mock = unittest.mock.Mock()
        rename_mock = unittest.mock.Mock()

        @unittest.mock.patch.object(service, "mkstemp", mktemp_mock)
        @unittest.mock.patch("os.symlink", symlink_mock)
        @unittest.mock.patch("os.rename", rename_mock)
        def _():
            self.service_manager.change_default_runlevel(runlevel)
            self.assertTrue(mktemp_mock.called)
            self.assertEqual(symlink_mock.call_args[0][0],  # pylint: disable=E1136
                             "/usr/lib/systemd/system/multi-user.target")
            self.assertEqual(rename_mock.call_args[0][1],  # pylint: disable=E1136
                             "/etc/systemd/system/default.target")
        _()

    def test_unknown_runlevel(self):
        self.assertRaises(ValueError,
                          service.convert_systemd_target_to_runlevel, "unknown")

    def test_runlevels(self):
        self.assertEqual(service.convert_sysv_runlevel(0), "poweroff.target")
        self.assertEqual(service.convert_sysv_runlevel(1), "rescue.target")
        self.assertEqual(service.convert_sysv_runlevel(2), "multi-user.target")
        self.assertEqual(service.convert_sysv_runlevel(5), "graphical.target")
        self.assertEqual(service.convert_sysv_runlevel(6), "reboot.target")


class TestSysVInitServiceManager(unittest.TestCase):

    def setUp(self):
        self.run_mock = unittest.mock.Mock()
        self.init_name = "init"
        self.service_manager = get_service_manager_from_init_and_run(
            self.init_name,
            self.run_mock)

    def test_start(self):
        srv = "lldpad"
        self.service_manager.start(srv)
        cmd = ("service %s start" % srv)
        self.assertEqual(self.run_mock.call_args[0][0], cmd)  # pylint: disable=E1136

    def test_list(self):
        list_result_mock = unittest.mock.Mock(
            exit_status=0,
            stdout_text="sshd             0:off   1:off   "
            "2:off   3:off   4:off   5:off   6:off\n"
            "vsftpd           0:off   1:off   2:off "
            "  3:off   4:off   5:on   6:off\n"
            "xinetd based services:\n"
            "        amanda:         off\n"
            "        chargen-dgram:  on\n")

        run_mock = unittest.mock.Mock(return_value=list_result_mock)
        service_manager = get_service_manager_from_init_and_run(self.init_name,
                                                                run_mock)
        list_result = service_manager.list(ignore_status=False)
        self.assertEqual(run_mock.call_args[0][0], "chkconfig --list")  # pylint: disable=E1136
        self.assertEqual(list_result,
                         {'sshd': {0: "off", 1: "off", 2: "off", 3: "off",
                                   4: "off", 5: "off", 6: "off"},
                          'vsftpd': {0: "off", 1: "off", 2: "off", 3: "off",
                                     4: "off", 5: "on", 6: "off"},
                          'xinetd': {'amanda': "off", 'chargen-dgram': "on"}})

    def test_enable(self):
        srv = "lldpad"
        self.service_manager.enable(srv)
        cmd = "chkconfig lldpad on"
        self.assertEqual(self.run_mock.call_args[0][0], cmd)  # pylint: disable=E1136

    def test_unknown_runlevel(self):
        self.assertRaises(ValueError,
                          service.convert_sysv_runlevel, "unknown")

    def test_runlevels(self):
        self.assertEqual(service.convert_systemd_target_to_runlevel(
            "poweroff.target"), '0')
        self.assertEqual(service.convert_systemd_target_to_runlevel(
            "rescue.target"), 's')
        self.assertEqual(service.convert_systemd_target_to_runlevel(
            "multi-user.target"), '3')
        self.assertEqual(service.convert_systemd_target_to_runlevel(
            "graphical.target"), '5')
        self.assertEqual(service.convert_systemd_target_to_runlevel(
            "reboot.target"), '6')


if __name__ == '__main__':
    unittest.main()