How to use check_exit_code method in avocado

Best Python code snippet using avocado_python Github


Full Screen

1# vim: tabstop=4 shiftwidth=4 softtabstop=42# Copyright 2011 OpenStack Foundation.3# All Rights Reserved.4#5# Licensed under the Apache License, Version 2.0 (the "License"); you may6# not use this file except in compliance with the License. You may obtain7# a copy of the License at8#9# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the14# License for the specific language governing permissions and limitations15# under the License.16"""17System-level utilities and helper functions.18"""19import os20import random21import shlex22import signal23from import subprocess24from eventlet import greenthread25from glance.openstack.common.gettextutils import _26from glance.openstack.common import log as logging27LOG = logging.getLogger(__name__)28class InvalidArgumentError(Exception):29 def __init__(self, message=None):30 super(InvalidArgumentError, self).__init__(message)31class UnknownArgumentError(Exception):32 def __init__(self, message=None):33 super(UnknownArgumentError, self).__init__(message)34class ProcessExecutionError(Exception):35 def __init__(self, stdout=None, stderr=None, exit_code=None, cmd=None,36 description=None):37 self.exit_code = exit_code38 self.stderr = stderr39 self.stdout = stdout40 self.cmd = cmd41 self.description = description42 if description is None:43 description = "Unexpected error while running command."44 if exit_code is None:45 exit_code = '-'46 message = ("%s\nCommand: %s\nExit code: %s\nStdout: %r\nStderr: %r"47 % (description, cmd, exit_code, stdout, stderr))48 super(ProcessExecutionError, self).__init__(message)49class NoRootWrapSpecified(Exception):50 def __init__(self, message=None):51 super(NoRootWrapSpecified, self).__init__(message)52def _subprocess_setup():53 # Python installs a SIGPIPE handler by default. This is usually not what54 # non-Python subprocesses expect.55 signal.signal(signal.SIGPIPE, signal.SIG_DFL)56def execute(*cmd, **kwargs):57 """58 Helper method to shell out and execute a command through subprocess with59 optional retry.60 :param cmd: Passed to subprocess.Popen.61 :type cmd: string62 :param process_input: Send to opened process.63 :type proces_input: string64 :param check_exit_code: Single bool, int, or list of allowed exit65 codes. Defaults to [0]. Raise66 :class:`ProcessExecutionError` unless67 program exits with one of these code.68 :type check_exit_code: boolean, int, or [int]69 :param delay_on_retry: True | False. Defaults to True. If set to True,70 wait a short amount of time before retrying.71 :type delay_on_retry: boolean72 :param attempts: How many times to retry cmd.73 :type attempts: int74 :param run_as_root: True | False. Defaults to False. If set to True,75 the command is prefixed by the command specified76 in the root_helper kwarg.77 :type run_as_root: boolean78 :param root_helper: command to prefix to commands called with79 run_as_root=True80 :type root_helper: string81 :param shell: whether or not there should be a shell used to82 execute this command. Defaults to false.83 :type shell: boolean84 :returns: (stdout, stderr) from process execution85 :raises: :class:`UnknownArgumentError` on86 receiving unknown arguments87 :raises: :class:`ProcessExecutionError`88 """89 process_input = kwargs.pop('process_input', None)90 check_exit_code = kwargs.pop('check_exit_code', [0])91 ignore_exit_code = False92 delay_on_retry = kwargs.pop('delay_on_retry', True)93 attempts = kwargs.pop('attempts', 1)94 run_as_root = kwargs.pop('run_as_root', False)95 root_helper = kwargs.pop('root_helper', '')96 shell = kwargs.pop('shell', False)97 if isinstance(check_exit_code, bool):98 ignore_exit_code = not check_exit_code99 check_exit_code = [0]100 elif isinstance(check_exit_code, int):101 check_exit_code = [check_exit_code]102 if kwargs:103 raise UnknownArgumentError(_('Got unknown keyword args '104 'to utils.execute: %r') % kwargs)105 if run_as_root and os.geteuid() != 0:106 if not root_helper:107 raise NoRootWrapSpecified(108 message=('Command requested root, but did not specify a root '109 'helper.'))110 cmd = shlex.split(root_helper) + list(cmd)111 cmd = map(str, cmd)112 while attempts > 0:113 attempts -= 1114 try:115 LOG.debug(_('Running cmd (subprocess): %s'), ' '.join(cmd))116 _PIPE = subprocess.PIPE # pylint: disable=E1101117 if == 'nt':118 preexec_fn = None119 close_fds = False120 else:121 preexec_fn = _subprocess_setup122 close_fds = True123 obj = subprocess.Popen(cmd,124 stdin=_PIPE,125 stdout=_PIPE,126 stderr=_PIPE,127 close_fds=close_fds,128 preexec_fn=preexec_fn,129 shell=shell)130 result = None131 if process_input is not None:132 result = obj.communicate(process_input)133 else:134 result = obj.communicate()135 obj.stdin.close() # pylint: disable=E1101136 _returncode = obj.returncode # pylint: disable=E1101137 if _returncode:138 LOG.debug(_('Result was %s') % _returncode)139 if not ignore_exit_code and _returncode not in check_exit_code:140 (stdout, stderr) = result141 raise ProcessExecutionError(exit_code=_returncode,142 stdout=stdout,143 stderr=stderr,144 cmd=' '.join(cmd))145 return result146 except ProcessExecutionError:147 if not attempts:148 raise149 else:150 LOG.debug(_('%r failed. Retrying.'), cmd)151 if delay_on_retry:152 greenthread.sleep(random.randint(20, 200) / 100.0)153 finally:154 # NOTE(termie): this appears to be necessary to let the subprocess155 # call clean something up in between calls, without156 # it two execute calls in a row hangs the second one157 greenthread.sleep(0)158def trycmd(*args, **kwargs):159 """160 A wrapper around execute() to more easily handle warnings and errors.161 Returns an (out, err) tuple of strings containing the output of162 the command's stdout and stderr. If 'err' is not empty then the163 command can be considered to have failed.164 :discard_warnings True | False. Defaults to False. If set to True,165 then for succeeding commands, stderr is cleared166 """167 discard_warnings = kwargs.pop('discard_warnings', False)168 try:169 out, err = execute(*args, **kwargs)170 failed = False171 except ProcessExecutionError, exn:172 out, err = '', str(exn)173 failed = True174 if not failed and discard_warnings and err:175 # Handle commands that output to stderr but otherwise succeed176 err = ''177 return out, err178def ssh_execute(ssh, cmd, process_input=None,179 addl_env=None, check_exit_code=True):180 LOG.debug(_('Running cmd (SSH): %s'), cmd)181 if addl_env:182 raise InvalidArgumentError(_('Environment not supported over SSH'))183 if process_input:184 # This is (probably) fixable if we need it...185 raise InvalidArgumentError(_('process_input not supported over SSH'))186 stdin_stream, stdout_stream, stderr_stream = ssh.exec_command(cmd)187 channel = stdout_stream.channel188 # NOTE(justinsb): This seems suspicious...189 # ...other SSH clients have buffering issues with this approach190 stdout = stderr = stdin_stream.close()193 exit_status = channel.recv_exit_status()194 # exit_status == -1 if no exit code was returned195 if exit_status != -1:196 LOG.debug(_('Result was %s') % exit_status)197 if check_exit_code and exit_status != 0:198 raise ProcessExecutionError(exit_code=exit_status,199 stdout=stdout,200 stderr=stderr,201 cmd=cmd)...

Full Screen

Full Screen Github


Full Screen

1# vim: tabstop=4 shiftwidth=4 softtabstop=42# Copyright 2013 OpenStack Foundation3# Copyright 2013 IBM Corp.4#5# Licensed under the Apache License, Version 2.0 (the "License"); you may6# not use this file except in compliance with the License. You may obtain7# a copy of the License at8#9# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the14# License for the specific language governing permissions and limitations15# under the License.16"""Provides methods needed by installation script for OpenStack development17virtual environments.18Since this script is used to bootstrap a virtualenv from the system's Python19environment, it should be kept strictly compatible with Python 2.6.20Synced in from openstack-common21"""22from __future__ import print_function23import optparse24import os25import subprocess26import sys27class InstallVenv(object):28 def __init__(self, root, venv, requirements,29 test_requirements, py_version,30 project):31 self.root = root32 self.venv = venv33 self.requirements = requirements34 self.test_requirements = test_requirements35 self.py_version = py_version36 self.project = project37 def die(self, message, *args):38 print(message % args, file=sys.stderr)39 sys.exit(1)40 def check_python_version(self):41 if sys.version_info < (2, 6):42 self.die("Need Python Version >= 2.6")43 def run_command_with_code(self, cmd, redirect_output=True,44 check_exit_code=True):45 """Runs a command in an out-of-process shell.46 Returns the output of that command. Working directory is self.root.47 """48 if redirect_output:49 stdout = subprocess.PIPE50 else:51 stdout = None52 proc = subprocess.Popen(cmd, cwd=self.root, stdout=stdout)53 output = proc.communicate()[0]54 if check_exit_code and proc.returncode != 0:55 self.die('Command "%s" failed.\n%s', ' '.join(cmd), output)56 return (output, proc.returncode)57 def run_command(self, cmd, redirect_output=True, check_exit_code=True):58 return self.run_command_with_code(cmd, redirect_output,59 check_exit_code)[0]60 def get_distro(self):61 if (os.path.exists('/etc/fedora-release') or62 os.path.exists('/etc/redhat-release')):63 return Fedora(64 self.root, self.venv, self.requirements,65 self.test_requirements, self.py_version, self.project)66 else:67 return Distro(68 self.root, self.venv, self.requirements,69 self.test_requirements, self.py_version, self.project)70 def check_dependencies(self):71 self.get_distro().install_virtualenv()72 def create_virtualenv(self, no_site_packages=True):73 """Creates the virtual environment and installs PIP.74 Creates the virtual environment and installs PIP only into the75 virtual environment.76 """77 if not os.path.isdir(self.venv):78 print('Creating venv...', end=' ')79 if no_site_packages:80 self.run_command(['virtualenv', '-q', '--no-site-packages',81 self.venv])82 else:83 self.run_command(['virtualenv', '-q', self.venv])84 print('done.')85 else:86 print("venv already exists...")87 pass88 def pip_install(self, *args):89 self.run_command(['tools/',90 'pip', 'install', '--upgrade'] + list(args),91 redirect_output=False)92 def install_dependencies(self):93 print('Installing dependencies with pip (this can take a while)...')94 # First things first, make sure our venv has the latest pip and95 # setuptools.96 self.pip_install('pip>=1.3')97 self.pip_install('setuptools')98 self.pip_install('-r', self.requirements)99 self.pip_install('-r', self.test_requirements)100 def post_process(self):101 self.get_distro().post_process()102 def parse_args(self, argv):103 """Parses command-line arguments."""104 parser = optparse.OptionParser()105 parser.add_option('-n', '--no-site-packages',106 action='store_true',107 help="Do not inherit packages from global Python "108 "install")109 return parser.parse_args(argv[1:])[0]110class Distro(InstallVenv):111 def check_cmd(self, cmd):112 return bool(self.run_command(['which', cmd],113 check_exit_code=False).strip())114 def install_virtualenv(self):115 if self.check_cmd('virtualenv'):116 return117 if self.check_cmd('easy_install'):118 print('Installing virtualenv via easy_install...', end=' ')119 if self.run_command(['easy_install', 'virtualenv']):120 print('Succeeded')121 return122 else:123 print('Failed')124 self.die('ERROR: virtualenv not found.\n\n%s development'125 ' requires virtualenv, please install it using your'126 ' favorite package management tool' % self.project)127 def post_process(self):128 """Any distribution-specific post-processing gets done here.129 In particular, this is useful for applying patches to code inside130 the venv.131 """132 pass133class Fedora(Distro):134 """This covers all Fedora-based distributions.135 Includes: Fedora, RHEL, CentOS, Scientific Linux136 """137 def check_pkg(self, pkg):138 return self.run_command_with_code(['rpm', '-q', pkg],139 check_exit_code=False)[1] == 0140 def apply_patch(self, originalfile, patchfile):141 self.run_command(['patch', '-N', originalfile, patchfile],142 check_exit_code=False)143 def install_virtualenv(self):144 if self.check_cmd('virtualenv'):145 return146 if not self.check_pkg('python-virtualenv'):147 self.die("Please install 'python-virtualenv'.")148 super(Fedora, self).install_virtualenv()149 def post_process(self):150 """Workaround for a bug in eventlet.151 This currently affects RHEL6.1, but the fix can safely be152 applied to all RHEL and Fedora distributions.153 This can be removed when the fix is applied upstream.154 Nova: Upstream: RHEL: """158 # Install "patch" program if it's not there159 if not self.check_pkg('patch'):160 self.die("Please install 'patch'.")161 # Apply the eventlet patch162 self.apply_patch(os.path.join(self.venv, 'lib', self.py_version,163 'site-packages',164 'eventlet/green/'),...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?