Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/graphviz/backend/execute.py: 56%

1"""Run subprocesses with ``subprocess.run()`` and ``subprocess.Popen()``.""" 

2 

3import errno 

4import logging 

5import os 

6import subprocess 

7import sys 

8import typing 

9 

10from .. import _compat 

11 

12__all__ = ['run_check', 'ExecutableNotFound', 'CalledProcessError'] 

13 

14 

15log = logging.getLogger(__name__) 

16 

17 

18BytesOrStrIterator = typing.Union[typing.Iterator[bytes], 

19 typing.Iterator[str]] 

20 

21 

22@typing.overload 

23def run_check(cmd: typing.Sequence[typing.Union[os.PathLike, str]], *, 

24 input_lines: typing.Optional[typing.Iterator[bytes]] = ..., 

25 encoding: None = ..., 

26 quiet: bool = ..., 

27 **kwargs) -> subprocess.CompletedProcess: 

28 """Accept bytes input_lines with default ``encoding=None```.""" 

29 

30 

31@typing.overload 

32def run_check(cmd: typing.Sequence[typing.Union[os.PathLike, str]], *, 

33 input_lines: typing.Optional[typing.Iterator[str]] = ..., 

34 encoding: str, 

35 quiet: bool = ..., 

36 **kwargs) -> subprocess.CompletedProcess: 

37 """Accept string input_lines when given ``encoding``.""" 

38 

39 

40@typing.overload 

41def run_check(cmd: typing.Sequence[typing.Union[os.PathLike, str]], *, 

42 input_lines: typing.Optional[BytesOrStrIterator] = ..., 

43 encoding: typing.Optional[str] = ..., 

44 capture_output: bool = ..., 

45 quiet: bool = ..., 

46 **kwargs) -> subprocess.CompletedProcess: 

47 """Accept bytes or string input_lines depending on ``encoding``.""" 

48 

49 

50def run_check(cmd: typing.Sequence[typing.Union[os.PathLike, str]], *, 

51 input_lines: typing.Optional[BytesOrStrIterator] = None, 

52 encoding: typing.Optional[str] = None, 

53 quiet: bool = False, 

54 **kwargs) -> subprocess.CompletedProcess: 

55 """Run the command described by ``cmd`` 

56 with ``check=True`` and return its completed process. 

57 

58 Raises: 

59 CalledProcessError: if the returncode of the subprocess is non-zero. 

60 """ 

61 log.debug('run %r', cmd) 

62 

63 cmd = list(map(_compat.make_subprocess_arg, cmd)) 

64 

65 if not kwargs.pop('check', True): # pragma: no cover 

66 raise NotImplementedError('check must be True or omited') 

67 

68 if encoding is not None: 

69 kwargs['encoding'] = encoding 

70 

71 kwargs.setdefault('startupinfo', _compat.get_startupinfo()) 

72 

73 try: 

74 if input_lines is not None: 

75 assert kwargs.get('input') is None 

76 assert iter(input_lines) is input_lines 

77 if kwargs.pop('capture_output'): 

78 kwargs['stdout'] = kwargs['stderr'] = subprocess.PIPE 

79 proc = _run_input_lines(cmd, input_lines, kwargs=kwargs) 

80 else: 

81 proc = subprocess.run(cmd, **kwargs) 

82 except OSError as e: 

83 if e.errno == errno.ENOENT: 

84 raise ExecutableNotFound(cmd) from e 

85 raise 

86 

87 if not quiet and proc.stderr: 

88 _write_stderr(proc.stderr) 

89 

90 try: 

91 proc.check_returncode() 

92 except subprocess.CalledProcessError as e: 

93 raise CalledProcessError(*e.args) 

94 

95 return proc 

96 

97 

98def _run_input_lines(cmd, input_lines, *, kwargs): 

99 popen = subprocess.Popen(cmd, stdin=subprocess.PIPE, **kwargs) 

100 

101 stdin_write = popen.stdin.write 

102 for line in input_lines: 

103 stdin_write(line) 

104 

105 stdout, stderr = popen.communicate() 

106 return subprocess.CompletedProcess(popen.args, popen.returncode, 

107 stdout=stdout, stderr=stderr) 

108 

109 

110def _write_stderr(stderr) -> None: 

111 if isinstance(stderr, bytes): 

112 stderr_encoding = (getattr(sys.stderr, 'encoding', None) 

113 or sys.getdefaultencoding()) 

114 stderr = stderr.decode(stderr_encoding) 

115 

116 sys.stderr.write(stderr) 

117 sys.stderr.flush() 

118 return None 

119 

120 

121class ExecutableNotFound(RuntimeError): 

122 """:exc:`RuntimeError` raised if the Graphviz executable is not found.""" 

123 

124 _msg = ('failed to execute {!r}, ' 

125 'make sure the Graphviz executables are on your systems\' PATH') 

126 

127 def __init__(self, args) -> None: 

128 super().__init__(self._msg.format(*args)) 

129 

130 

131class CalledProcessError(subprocess.CalledProcessError): 

132 """:exc:`~subprocess.CalledProcessError` raised if a subprocess ``returncode`` is not ``0``.""" # noqa: E501 

133 

134 def __str__(self) -> 'str': 

135 return f'{super().__str__()} [stderr: {self.stderr!r}]'