Repository: firecracker-microvm/firecracker
File: tests/framework/utils.py (Python)

Code before:
    :param cwd: sets the current directory before the child is executed
    :param timeout: Time before command execution should be aborted with a `TimeoutExpired` exception
    :param warn_on_fail: Whether to log a warning if an error occurs. Turn off if failures are expected and retried,
                         or logged at a higher level already, to avoid red herrings in logs.
    :return: return code, stdout, stderr
    """
    if isinstance(cmd, list) or not shell:
        # Create the async process
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
        )
    else:
        proc = subprocess.Popen(
            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
        )

    try:
        stdout, stderr = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        stdout, stderr = proc.communicate()

        if warn_on_fail:
            # Log the message with one call so that multiple statuses
            # don't get mixed up
            CMDLOG.warning(
                "Timeout executing command: %s\n",
                _format_output_message(proc, stdout, stderr),
            )

        raise

    output_message = _format_output_message(proc, stdout, stderr)

    # If a non-zero return code was thrown, raise an exception
    if check and proc.returncode != 0:
        if warn_on_fail:
            CMDLOG.warning("Command failed: %s\n", output_message)

        raise ChildProcessError(output_message)

    CMDLOG.debug(output_message)

    return CommandReturn(proc.returncode, stdout.decode(), stderr.decode())


def check_output(
    cmd, shell=True, cwd=None, timeout=None, warn_on_fail=True
) -> CommandReturn:
    """Identical to `run_cmd`, but always sets `check_output` to `True`."""
    return run_cmd(cmd, True, shell, cwd, timeout, warn_on_fail)

Review comment (suggestion):
Maybe instead of warning we should attach proc, stdout and stderr to the exception and raise that. The caller can choose to suppress it or display it.

Code after:
    :return: return code, stdout, stderr
    """
    if isinstance(cmd, list) or not shell:
        # Create the async process
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
        )
    else:
        proc = subprocess.Popen(
            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
        )

    try:
        stdout, stderr = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        stdout, stderr = proc.communicate()

        # Log the message with one call so that multiple statuses
        # don't get mixed up
        CMDLOG.warning(
            "Timeout executing command: %s\n",
            _format_output_message(proc, stdout, stderr),
        )

        raise

    output_message = _format_output_message(proc, stdout, stderr)

    # If a non-zero return code was thrown, raise an exception
    if check and proc.returncode != 0:
        CMDLOG.warning("Command failed: %s\n", output_message)

        raise ChildProcessError(output_message)

    CMDLOG.debug(output_message)

    return CommandReturn(proc.returncode, stdout.decode(), stderr.decode())


def check_output(cmd, shell=True, cwd=None, timeout=None) -> CommandReturn:
    """Identical to `run_cmd`, but always sets `check_output` to `True`."""
    return run_cmd(cmd, True, shell, cwd, timeout)


def assert_seccomp_level(pid, seccomp_level):
    """Test that seccomp_level applies to all threads of a process."""
    # Get number of threads
    cmd = "ps -T --no-headers -p {} | awk '{{print $2}}'".format(pid)
    process = check_output(cmd)
    threads_out_lines = process.stdout.splitlines()