How to use specific_service_manager method in avocado

Best Python code snippet using avocado_python

service.py

Source:service.py

...628    # remove list method629    command_list = [(c, r) for (c, r) in COMMANDS if630                    c not in ["list", "set_target"]]631    return _ServiceCommandGenerator(command_generator, command_list)632def specific_service_manager(service_name, run=process.run):633    """Get the service manager for a specific service.634    Example of use:635    .. code-block:: python636      # Get the specific service manager for sshd637      sshd = SpecificServiceManager("sshd")638      sshd.start()639      sshd.stop()640      sshd.reload()641      sshd.restart()642      sshd.condrestart()643      sshd.status()644      sshd.enable()645      sshd.disable()646      sshd.is_enabled()...

service_lib.py

Source:service_lib.py

1# Copyright (c) 2013 - 2017, Intel Corporation.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#    http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""``service_lib.py``15`SystemD service library`16"""17import os18from tempfile import mktemp19USAGE = """20import service_lib21# to generate raw list of str commands, e.g.  ["systemctl", "start", "networkd"]22systemd_cmd_gen = service_lib.systemd_command_generator()23cmd = " ".join(systemd_cmd_gen.start("networkd"))24# to directly call cli_send_command but still accepts kwargs for cli_send_command25networkd = service_lib.SpecificServiceManager("networkd", self.cli_send_command)26networkd.stop(expected_rcs={0, 3})27networkd.start(expected_rcs={1})28# to directly call cli_send_command but still accepts kwargs for cli_send_command29systemd = service_lib.systemd_manager_factory(self.cli_send_command)30systemd.stop("networkd", expected_rcs={0, 3})31systemd.list(expected_rcs={0, 3})32systemd.start(service="networkd", expected_rcs={0, 3})33systemd.change_default_runlevel("rescue.target")34"""35COMMAND_TABLE_DOC = """36service frobozz start37systemctl start frobozz.service38 Used to start a service (not reboot persistent)39service frobozz stop40systemctl stop frobozz.service41 Used to stop a service (not reboot persistent)42service frobozz restart43systemctl restart frobozz.service44 Used to stop and then start a service45service frobozz reload46systemctl reload frobozz.service47 When supported, reloads the config file without interrupting pending operations.48service frobozz condrestart49systemctl condrestart frobozz.service50 Restarts if the service is already running.51service frobozz status52systemctl status frobozz.service53 Tells whether a service is currently running.54ls /etc/rc.d/init.d/55systemctl list-unit-files --type=service (preferred)56 Used to list the services that can be started or stopped57ls /lib/systemd/system/*.service /etc/systemd/system/*.service58 Used to list all the services and other units59chkconfig frobozz on60systemctl enable frobozz.service61 Turn the service on, for start at next boot, or other trigger.62chkconfig frobozz off63systemctl disable frobozz.service64 Turn the service off for the next reboot, or any other trigger.65chkconfig frobozz66systemctl is-enabled frobozz.service67 Used to check whether a service is configured to start or not in the current environment.68chkconfig --list69systemctl list-unit-files --type=service(preferred)70ls /etc/systemd/system/*.wants/71 Print a table of services that lists which runlevels each is configured on or off72chkconfig frobozz --list73ls /etc/systemd/system/*.wants/frobozz.service74 Used to list what levels this service is configured on or off75chkconfig frobozz --add76systemctl daemon-reload77 Used when you create a new service file or modify any configuration78"""79class ReturnCodes(object):80    SUCCESS = 081    RUNNING = 082    STOPPED = 383    UNKNOWN = None84class SystemdReturnCodes(ReturnCodes):85    pass86REPLACE_COMMAND_LIST = {87    'is_enabled',88    'is_active',89    'daemon_reload',90}91COMMANDS = {92    "start",93    "stop",94    "reload",95    "restart",96    "condrestart",97    "status",98    "enable",99    "disable",100    "is_enabled",101    "is_active",102    "list",103    "daemon_reload",104    "cat"105}106def systemd_command_generator(command):107    command_name = "systemctl"108    if command in REPLACE_COMMAND_LIST:109        command = command.replace('_', '-')110    if command == "list":111        # noinspection PyUnusedLocal112        def list_command(_):113            return [command_name, "list-unit-files", "--type=service"]114        return list_command115    elif command == "daemon-reload":116        def daemon_reload_command(*_):117            return [command_name, command, '']118        return daemon_reload_command119    def method(service_name):120        return [command_name, command, "{}.service".format(service_name)]121    return method122class ServiceCommandGenerator(object):123    def __getattr__(self, name):124        if name not in self:125            raise AttributeError(name)126        command = self.command_generator(name)127        setattr(self, name, command)128        return command129    def __iter__(self):130        return iter(self.commands)131    def __contains__(self, value):132        return value in self.commands133    def __init__(self, command_generator, return_codes=ReturnCodes, command_list=None):134        super(ServiceCommandGenerator, self).__init__()135        if command_list is None:136            command_list = COMMANDS137        self.commands = command_list138        self.command_generator = command_generator139        self.return_codes = return_codes140class GenericServiceManager(object):141    def __init__(self, run_func, command_list=None):142        super().__init__()143        if command_list is None:144            command_list = COMMANDS145        self.service_command_generator = ServiceCommandGenerator(systemd_command_generator,146                                                                 SystemdReturnCodes,147                                                                 command_list)148        self.return_codes = SystemdReturnCodes149        self.run_func = run_func150    def __getattr__(self, name):151        def run(service='', **kwargs):152            return self.run_func(' '.join(command(service)), **kwargs)153        command = getattr(self.service_command_generator, name)154        setattr(self, name, run)155        return run156    def _get_running_status(self, service=''):157        return self.status(service=service, expected_rcs={self.return_codes.RUNNING,158                                                          self.return_codes.STOPPED})159    def is_running(self, service=''):160        rv = self._get_running_status(service)161        return rv.rc == self.return_codes.RUNNING162    def is_stopped(self, service=''):163        rv = self._get_running_status(service)164        return rv.rc == self.return_codes.STOPPED165class SpecificServiceManager(GenericServiceManager):166    def __init__(self, service_name, run_func):167        command_list = [c for c in COMMANDS if c != "list"]168        super().__init__(run_func, command_list)169        self.service_name = service_name170    def __getattr__(self, name):171        def run(**kwargs):172            kwargs.pop('service', None)  # remove any value associated with the service key173            return self.run_func(command, **kwargs)174        command = getattr(self.service_command_generator, name)175        command = ' '.join(command(self.service_name))176        setattr(self, name, run)177        return run178class SystemdServiceManager(GenericServiceManager):179    def __init__(self, run):180        super().__init__(run)181    @staticmethod182    def change_default_runlevel(runlevel='multi-user.target'):183        # atomic symlinking, symlink and then rename184        tmp_symlink = mktemp(dir="/etc/systemd/system")185        os.symlink("/usr/lib/systemd/system/{}".format(runlevel), tmp_symlink)186        os.rename(tmp_symlink, "/etc/systemd/system/default.target")187class ServiceConfigChangeContext(object):188    """Context manager suitable for service configuration.189    """190    def __init__(self, specific_service_manager):191        super().__init__()192        self.rcs = specific_service_manager.return_codes193        self.was_running = None194        self.specific_service_manager = specific_service_manager195    def __enter__(self):196        self.was_running = self.specific_service_manager.is_running()197        if self.was_running:198            self.specific_service_manager.stop()199    def __exit__(self, exc_type, exc, exc_tb):200        self.specific_service_manager.daemon_reload()201        if self.was_running:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.