Best Python code snippet using avocado_python
service.py
Source:service.py
...322 """323 self.commands = command_list324 for command in self.commands:325 setattr(self, command, command_generator(command))326def _get_name_of_init(run=utils.run):327 """328 Internal function to determine what executable is PID 1,329 aka init by checking /proc/1/exe330 :return: executable name for PID 1, aka init331 :rtype: str332 """333 # /proc/1/comm was added in 2.6.33 and is not in RHEL6.x, so use cmdline334 # Non-root can read cmdline335 # return os.path.basename(open("/proc/1/cmdline").read().split(chr(0))[0])336 # readlink /proc/1/exe requires root337 # inspired by openvswitch.py:ServiceManagerInterface.get_version()338 output = run("readlink /proc/1/exe").stdout.strip()339 return os.path.basename(output)340def get_name_of_init(run=utils.run):341 """342 Determine what executable is PID 1, aka init by checking /proc/1/exe343 This init detection will only run once and cache the return value.344 :return: executable name for PID 1, aka init345 :rtype: str346 """347 # _init_name is explicitly undefined so that we get the NameError on first access348 # pylint: disable=W0601349 global _init_name350 try:351 return _init_name352 except (NameError, AttributeError):353 _init_name = _get_name_of_init(run)354 return _init_name355class _SpecificServiceManager(object):356 def __init__(self, service_name, service_command_generator, service_result_parser, run=utils.run):357 """358 Create staticmethods that call utils.run with the given service_name359 for each command in service_command_generator.360 lldpad = SpecificServiceManager("lldpad",361 auto_create_specific_service_command_generator())362 lldpad.start()363 lldpad.stop()364 :param service_name: init service name or systemd unit name365 :type service_name: str366 :param service_command_generator: a sys_v_init or systemd command generator367 :type service_command_generator: _ServiceCommandGenerator368 :param run: function that executes the commands and return CmdResult object, default utils.run369 :type run: function370 """371 for cmd in service_command_generator.commands:372 setattr(self, cmd,373 self.generate_run_function(run,374 getattr(375 service_result_parser, cmd),376 getattr(377 service_command_generator, cmd),378 service_name))379 @staticmethod380 def generate_run_function(run_func, parse_func, command, service_name):381 """382 Generate the wrapped call to utils.run for the given service_name.383 :param run_func: function to execute command and return CmdResult object.384 :type run_func: function385 :param parse_func: function to parse the result from run.386 :type parse_func: function387 :param command: partial function that generates the command list388 :type command: function389 :param service_name: init service name or systemd unit name390 :type service_name: str391 :return: wrapped utils.run function.392 :rtype: function393 """394 def run(**kwargs):395 """396 Wrapped utils.run invocation that will start, stop, restart, etc. a service.397 :param kwargs: extra arguments to utils.run, .e.g. timeout. But not for ignore_status.398 We need a CmdResult to parse and raise a error.TestError if command failed.399 We will not let the CmdError out.400 :return: result of parse_func.401 """402 # If run_func is utils.run by default, we need to set403 # ignore_status = True. Otherwise, skip this setting.404 if run_func is utils.run:405 logging.debug("Setting ignore_status to True.")406 kwargs["ignore_status"] = True407 result = run_func(" ".join(command(service_name)), **kwargs)408 return parse_func(result)409 return run410class _GenericServiceManager(object):411 """412 Base class for SysVInitServiceManager and SystemdServiceManager.413 """414 def __init__(self, service_command_generator, service_result_parser, run=utils.run):415 """416 Create staticmethods for each service command, e.g. start, stop, restart.417 These staticmethods take as an argument the service to be started or stopped.418 systemd = SpecificServiceManager(auto_create_specific_service_command_generator())419 systemd.start("lldpad")420 systemd.stop("lldpad")421 :param service_command_generator: a sys_v_init or systemd command generator422 :type service_command_generator: _ServiceCommandGenerator423 :param run: function to call the run the commands, default utils.run424 :type run: function425 """426 # create staticmethods in class attributes (not used)427 # for cmd in service_command_generator.commands:428 # setattr(self.__class__, cmd,429 # staticmethod(self.generate_run_function(run, getattr(service_command_generator, cmd))))430 # create functions in instance attributes431 for cmd in service_command_generator.commands:432 setattr(self, cmd,433 self.generate_run_function(run,434 getattr(435 service_result_parser, cmd),436 getattr(service_command_generator, cmd)))437 @staticmethod438 def generate_run_function(run_func, parse_func, command):439 """440 Generate the wrapped call to utils.run for the service command, "service" or "systemctl"441 :param run_func: utils.run442 :type run_func: function443 :param command: partial function that generates the command list444 :type command: function445 :return: wrapped utils.run function.446 :rtype: function447 """448 def run(service="", **kwargs):449 """450 Wrapped utils.run invocation that will start, stop, restart, etc. a service.451 :param service: service name, e.g. crond, dbus, etc.452 :param kwargs: extra arguments to utils.run, .e.g. timeout. But not for ignore_status.453 We need a CmdResult to parse and raise a error.TestError if command failed.454 We will not let the CmdError out.455 :return: result of parse_func.456 """457 # If run_func is utils.run by default, we need to set458 # ignore_status = True. Otherwise, skip this setting.459 if run_func is utils.run:460 logging.debug("Setting ignore_status to True.")461 kwargs["ignore_status"] = True462 result = run_func(" ".join(command(service)), **kwargs)463 return parse_func(result)464 return run465class _SysVInitServiceManager(_GenericServiceManager):466 """467 Concrete class that implements the SysVInitServiceManager468 """469 def __init__(self, service_command_generator, service_result_parser, run=utils.run):470 """471 Create the GenericServiceManager for SysV services.472 :param service_command_generator:473 :type service_command_generator: _ServiceCommandGenerator474 :param run: function to call to run the commands, default utils.run475 :type run: function476 """477 super(478 _SysVInitServiceManager, self).__init__(service_command_generator,479 service_result_parser, run)480 # @staticmethod481 # def change_default_runlevel(runlevel='3'):482 # """483 # Set the default sys_v runlevel484 #485 # :param runlevel: sys_v runlevel to set as default in inittab486 # :type runlevel: str487 # """488 # raise NotImplemented489def convert_sysv_runlevel(level):490 """491 Convert runlevel to systemd target.492 :param level: sys_v runlevel493 :type level: str or int494 :return: systemd target495 :rtype: str496 :raise ValueError: when runlevel is unknown497 """498 runlevel = str(level)499 if runlevel == '0':500 target = "poweroff.target"501 elif runlevel in ['1', "s", "single"]:502 target = "rescue.target"503 elif runlevel in ['2', '3', '4']:504 target = "multi-user.target"505 elif runlevel == '5':506 target = "graphical.target"507 elif runlevel == '6':508 target = "reboot.target"509 else:510 raise ValueError("unknown runlevel %s" % level)511 return target512def convert_systemd_target_to_runlevel(target):513 """514 Convert systemd target to runlevel.515 :param target: systemd target516 :type target: str517 :return: sys_v runlevel518 :rtype: str519 :raise ValueError: when systemd target is unknown520 """521 if target == "poweroff.target":522 runlevel = '0'523 elif target == "rescue.target":524 runlevel = 's'525 elif target == "multi-user.target":526 runlevel = '3'527 elif target == "graphical.target":528 runlevel = '5'529 elif target == "reboot.target":530 runlevel = '6'531 else:532 raise ValueError("unknown target %s" % target)533 return runlevel534class _SystemdServiceManager(_GenericServiceManager):535 """536 Concrete class that implements the SystemdServiceManager537 """538 def __init__(self, service_command_generator, service_result_parser, run=utils.run):539 """540 Create the GenericServiceManager for systemd services.541 :param service_command_generator:542 :type service_command_generator: _ServiceCommandGenerator543 :param run: function to call to run the commands, default utils.run544 :type run: function545 """546 super(_SystemdServiceManager, self).__init__(service_command_generator,547 service_result_parser, run)548 @staticmethod549 def change_default_runlevel(runlevel='multi-user.target'):550 # atomic symlinking, symlink and then rename551 """552 Set the default systemd target.553 Create the symlink in a temp directory and then use554 atomic rename to move the symlink into place.555 :param runlevel: default systemd target556 :type runlevel: str557 """558 tmp_symlink = mktemp(dir="/etc/systemd/system")559 os.symlink("/usr/lib/systemd/system/%s" % runlevel, tmp_symlink)560 os.rename(tmp_symlink, "/etc/systemd/system/default.target")561_command_generators = {"init": sys_v_init_command_generator,562 "systemd": systemd_command_generator}563_result_parsers = {"init": sys_v_init_result_parser,564 "systemd": systemd_result_parser}565_service_managers = {"init": _SysVInitServiceManager,566 "systemd": _SystemdServiceManager}567def _get_service_result_parser(run=utils.run):568 """569 Get the ServiceResultParser using the auto-detect init command.570 :return: ServiceResultParser fro the current init command.571 :rtype: _ServiceResultParser572 """573 # pylint: disable=W0601574 global _service_result_parser575 try:576 return _service_result_parser577 except NameError:578 result_parser = _result_parsers[get_name_of_init(run)]579 _service_result_parser = _ServiceResultParser(result_parser)580 return _service_result_parser581def _get_service_command_generator(run=utils.run):582 """583 Lazy initializer for ServiceCommandGenerator using the auto-detect init command.584 :return: ServiceCommandGenerator for the current init command.585 :rtype: _ServiceCommandGenerator586 """587 # _service_command_generator is explicitly undefined so that we get the NameError on first access588 # pylint: disable=W0601589 global _service_command_generator590 try:591 return _service_command_generator592 except NameError:593 command_generator = _command_generators[get_name_of_init(run)]594 _service_command_generator = _ServiceCommandGenerator(595 command_generator)596 return _service_command_generator597def ServiceManager(run=utils.run):598 """599 Detect which init program is being used, init or systemd and return a600 class has methods to start/stop services.601 # Get the system service manager602 service_manager = ServiceManager()603 # Stating service/unit "sshd"604 service_manager.start("sshd")605 # Getting a list of available units606 units = service_manager.list()607 # Disabling and stopping a list of services608 services_to_disable = ['ntpd', 'httpd']609 for s in services_to_disable:610 service_manager.disable(s)611 service_manager.stop(s)612 :return: SysVInitServiceManager or SystemdServiceManager613 :rtype: _GenericServiceManager614 """615 service_manager = _service_managers[get_name_of_init(run)]616 # _service_command_generator is explicitly undefined so that we get the NameError on first access617 # pylint: disable=W0601618 global _service_manager619 try:620 return _service_manager621 except NameError:622 _service_manager = service_manager(_get_service_command_generator(run),623 _get_service_result_parser(run))624 return _service_manager625def _auto_create_specific_service_result_parser(run=utils.run):626 """627 Create a class that will create partial functions that generate result_parser628 for the current init command.629 :return: A ServiceResultParser for the auto-detected init command.630 :rtype: _ServiceResultParser631 """632 result_parser = _result_parsers[get_name_of_init(run)]633 # remove list method634 command_list = [c for c in COMMANDS if c not in ["list", "set_target"]]635 return _ServiceResultParser(result_parser, command_list)636def _auto_create_specific_service_command_generator(run=utils.run):637 """638 Create a class that will create partial functions that generate commands639 for the current init command.640 lldpad = SpecificServiceManager("lldpad",641 auto_create_specific_service_command_generator())642 lldpad.start()643 lldpad.stop()644 :return: A ServiceCommandGenerator for the auto-detected init command.645 :rtype: _ServiceCommandGenerator646 """647 command_generator = _command_generators[get_name_of_init(run)]648 # remove list method649 command_list = [c for c in COMMANDS if c not in ["list", "set_target"]]650 return _ServiceCommandGenerator(command_generator, command_list)651def SpecificServiceManager(service_name, run=utils.run):652 """653 # Get the specific service manager for sshd654 sshd = SpecificServiceManager("sshd")655 sshd.start()656 sshd.stop()657 sshd.reload()658 sshd.restart()659 sshd.condrestart()660 sshd.status()661 sshd.enable()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!