def aws()

in awscli/testutils.py [0:0]

• 36 lines of code
• 11 McCabe index (conditional complexity)

def aws(command, collect_memory=False, env_vars=None,
        wait_for_finish=True, input_data=None, input_file=None):
    """Run an aws command.

    This help function abstracts the differences of running the "aws"
    command on different platforms.

    If collect_memory is ``True`` the the Result object will have a list
    of memory usage taken at 2 second intervals.  The memory usage
    will be in bytes.

    If env_vars is None, this will set the environment variables
    to be used by the aws process.

    If wait_for_finish is False, then the Process object is returned
    to the caller.  It is then the caller's responsibility to ensure
    proper cleanup.  This can be useful if you want to test timeout's
    or how the CLI responds to various signals.

    :type input_data: string
    :param input_data: This string will be communicated to the process through
        the stdin of the process.  It essentially allows the user to
        avoid having to use a file handle to pass information to the process.
        Note that this string is not passed on creation of the process, but
        rather communicated to the process.

    :type input_file: a file handle
    :param input_file: This is a file handle that will act as the
        the stdin of the process immediately on creation.  Essentially
        any data written to the file will be read from stdin of the
        process. This is needed if you plan to stream data into stdin while
        collecting memory.
    """
    if platform.system() == 'Windows':
        command = _escape_quotes(command)
    if 'AWS_TEST_COMMAND' in os.environ:
        aws_command = os.environ['AWS_TEST_COMMAND']
    else:
        aws_command = 'python %s' % get_aws_cmd()
    full_command = '%s %s' % (aws_command, command)
    stdout_encoding = get_stdout_encoding()
    if isinstance(full_command, six.text_type) and not six.PY3:
        full_command = full_command.encode(stdout_encoding)
    INTEG_LOG.debug("Running command: %s", full_command)
    env = os.environ.copy()
    if 'AWS_DEFAULT_REGION' not in env:
        env['AWS_DEFAULT_REGION'] = "us-east-1"
    if env_vars is not None:
        env = env_vars
    if input_file is None:
        input_file = PIPE
    process = Popen(full_command, stdout=PIPE, stderr=PIPE, stdin=input_file,
                    shell=True, env=env)
    if not wait_for_finish:
        return process
    memory = None
    if not collect_memory:
        kwargs = {}
        if input_data:
            kwargs = {'input': input_data}
        stdout, stderr = process.communicate(**kwargs)
    else:
        stdout, stderr, memory = _wait_and_collect_mem(process)
    return Result(process.returncode,
                  stdout.decode(stdout_encoding),
                  stderr.decode(stdout_encoding),
                  memory)