File: command_runner.py

package info (click to toggle)
diff-cover 10.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,252 kB
  • sloc: python: 6,425; xml: 218; cpp: 18; sh: 12; makefile: 10
file content (65 lines) | stat: -rw-r--r-- 1,860 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import subprocess
import sys


class CommandError(Exception):
    """
    Error raised when a command being executed returns an error
    """


def execute(command, exit_codes=None):
    """Execute provided command returning the stdout
    Args:
        command (list[str]): list of tokens to execute as your command.
        exit_codes (list[int]): exit codes which do not indicate error.
        subprocess_mod (module): Defaults to pythons subprocess module but you can optionally pass
        in another. This is mostly for testing purposes
    Returns:
        str - Stdout of the command passed in. This will be Unicode for python < 3. Str for python 3
    Raises:
        ValueError if there is a error running the command
    """
    if exit_codes is None:
        exit_codes = [0]

    stdout_pipe = subprocess.PIPE
    with subprocess.Popen(command, stdout=stdout_pipe, stderr=stdout_pipe) as process:
        try:
            stdout, stderr = process.communicate()
        except OSError:
            sys.stderr.write(" ".join(_ensure_unicode(cmd) for cmd in command))
            raise

    stderr = _ensure_unicode(stderr)
    if process.returncode not in exit_codes:
        raise CommandError(stderr)

    return _ensure_unicode(stdout), stderr


def run_command_for_code(command):
    """
    Returns command's exit code.
    """
    try:
        with subprocess.Popen(
            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        ) as process:
            process.communicate()
    except FileNotFoundError:
        return 1
    return process.returncode


def _ensure_unicode(text):
    """
    Ensures the text passed in becomes unicode
    Args:
        text (str|unicode)
    Returns:
        unicode
    """
    if isinstance(text, bytes):
        return text.decode(sys.getfilesystemencoding(), "replace")
    return text