Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/_core/_subprocesses.py: 26%

1from io import BytesIO 

2from os import PathLike 

3from subprocess import DEVNULL, PIPE, CalledProcessError, CompletedProcess 

4from typing import ( 

5 IO, 

6 Any, 

7 AsyncIterable, 

8 List, 

9 Mapping, 

10 Optional, 

11 Sequence, 

12 Union, 

13 cast, 

14) 

15 

16from ..abc import Process 

17from ._eventloop import get_asynclib 

18from ._tasks import create_task_group 

19 

20 

21async def run_process( 

22 command: Union[str, bytes, Sequence[Union[str, bytes]]], 

23 *, 

24 input: Optional[bytes] = None, 

25 stdout: Union[int, IO[Any], None] = PIPE, 

26 stderr: Union[int, IO[Any], None] = PIPE, 

27 check: bool = True, 

28 cwd: Union[str, bytes, "PathLike[str]", None] = None, 

29 env: Optional[Mapping[str, str]] = None, 

30 start_new_session: bool = False, 

31) -> "CompletedProcess[bytes]": 

32 """ 

33 Run an external command in a subprocess and wait until it completes. 

34 

35 .. seealso:: :func:`subprocess.run` 

36 

37 :param command: either a string to pass to the shell, or an iterable of strings containing the 

38 executable name or path and its arguments 

39 :param input: bytes passed to the standard input of the subprocess 

40 :param stdout: either :data:`subprocess.PIPE` or :data:`subprocess.DEVNULL` 

41 :param stderr: one of :data:`subprocess.PIPE`, :data:`subprocess.DEVNULL` or 

42 :data:`subprocess.STDOUT` 

43 :param check: if ``True``, raise :exc:`~subprocess.CalledProcessError` if the process 

44 terminates with a return code other than 0 

45 :param cwd: If not ``None``, change the working directory to this before running the command 

46 :param env: if not ``None``, this mapping replaces the inherited environment variables from the 

47 parent process 

48 :param start_new_session: if ``true`` the setsid() system call will be made in the child 

49 process prior to the execution of the subprocess. (POSIX only) 

50 :return: an object representing the completed process 

51 :raises ~subprocess.CalledProcessError: if ``check`` is ``True`` and the process exits with a 

52 nonzero return code 

53 

54 """ 

55 

56 async def drain_stream(stream: AsyncIterable[bytes], index: int) -> None: 

57 buffer = BytesIO() 

58 async for chunk in stream: 

59 buffer.write(chunk) 

60 

61 stream_contents[index] = buffer.getvalue() 

62 

63 async with await open_process( 

64 command, 

65 stdin=PIPE if input else DEVNULL, 

66 stdout=stdout, 

67 stderr=stderr, 

68 cwd=cwd, 

69 env=env, 

70 start_new_session=start_new_session, 

71 ) as process: 

72 stream_contents: List[Optional[bytes]] = [None, None] 

73 try: 

74 async with create_task_group() as tg: 

75 if process.stdout: 

76 tg.start_soon(drain_stream, process.stdout, 0) 

77 if process.stderr: 

78 tg.start_soon(drain_stream, process.stderr, 1) 

79 if process.stdin and input: 

80 await process.stdin.send(input) 

81 await process.stdin.aclose() 

82 

83 await process.wait() 

84 except BaseException: 

85 process.kill() 

86 raise 

87 

88 output, errors = stream_contents 

89 if check and process.returncode != 0: 

90 raise CalledProcessError(cast(int, process.returncode), command, output, errors) 

91 

92 return CompletedProcess(command, cast(int, process.returncode), output, errors) 

93 

94 

95async def open_process( 

96 command: Union[str, bytes, Sequence[Union[str, bytes]]], 

97 *, 

98 stdin: Union[int, IO[Any], None] = PIPE, 

99 stdout: Union[int, IO[Any], None] = PIPE, 

100 stderr: Union[int, IO[Any], None] = PIPE, 

101 cwd: Union[str, bytes, "PathLike[str]", None] = None, 

102 env: Optional[Mapping[str, str]] = None, 

103 start_new_session: bool = False, 

104) -> Process: 

105 """ 

106 Start an external command in a subprocess. 

107 

108 .. seealso:: :class:`subprocess.Popen` 

109 

110 :param command: either a string to pass to the shell, or an iterable of strings containing the 

111 executable name or path and its arguments 

112 :param stdin: one of :data:`subprocess.PIPE`, :data:`subprocess.DEVNULL`, a 

113 file-like object, or ``None`` 

114 :param stdout: one of :data:`subprocess.PIPE`, :data:`subprocess.DEVNULL`, 

115 a file-like object, or ``None`` 

116 :param stderr: one of :data:`subprocess.PIPE`, :data:`subprocess.DEVNULL`, 

117 :data:`subprocess.STDOUT`, a file-like object, or ``None`` 

118 :param cwd: If not ``None``, the working directory is changed before executing 

119 :param env: If env is not ``None``, it must be a mapping that defines the environment 

120 variables for the new process 

121 :param start_new_session: if ``true`` the setsid() system call will be made in the child 

122 process prior to the execution of the subprocess. (POSIX only) 

123 :return: an asynchronous process object 

124 

125 """ 

126 shell = isinstance(command, str) 

127 return await get_asynclib().open_process( 

128 command, 

129 shell=shell, 

130 stdin=stdin, 

131 stdout=stdout, 

132 stderr=stderr, 

133 cwd=cwd, 

134 env=env, 

135 start_new_session=start_new_session, 

136 )