Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use check_exit_code method in avocado

Best Python code snippet using avocado_python

processutils.py

Source:processutils.py

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#         http://www.apache.org/licenses/LICENSE-2.010#11#    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 eventlet.green 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 os.name == '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 = stdout_stream.read()191    stderr = stderr_stream.read()192    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)...

install_venv_common.py

Source:install_venv_common.py

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#         http://www.apache.org/licenses/LICENSE-2.010#11#    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/with_venv.sh',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: https://bugs.launchpad.net/nova/+bug/884915155        Upstream: https://bitbucket.org/eventlet/eventlet/issue/89156        RHEL: https://bugzilla.redhat.com/958868157        """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/subprocess.py'),...

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.