Coverage for src / competitive_verifier / exec.py: 100%

1import os 

2import subprocess 

3import sys 

4from contextlib import nullcontext 

5from logging import getLogger 

6from typing import TYPE_CHECKING, Literal, Optional, overload 

7 

8from competitive_verifier import log 

9 

10if TYPE_CHECKING: 

11 from _typeshed import StrOrBytesPath 

12 

13 _StrOrListStr = str | list[str] 

14 

15 

16logger = getLogger(__name__) 

17 

18 

19@overload 

20def exec_command( 

21 command: "_StrOrListStr", 

22 *, 

23 text: Literal[False] = False, 

24 check: bool = False, 

25 capture_output: bool = False, 

26 env: dict[str, str] | None = None, 

27 cwd: Optional["StrOrBytesPath"] = None, 

28 group_log: bool = False, 

29) -> subprocess.CompletedProcess[bytes]: ... 

30 

31 

32@overload 

33def exec_command( 

34 command: "_StrOrListStr", 

35 *, 

36 text: Literal[True], 

37 check: bool = False, 

38 capture_output: bool = False, 

39 env: dict[str, str] | None = None, 

40 cwd: Optional["StrOrBytesPath"] = None, 

41 group_log: bool = False, 

42) -> subprocess.CompletedProcess[str]: ... 

43 

44 

45@overload 

46def exec_command( 

47 command: "_StrOrListStr", 

48 *, 

49 text: bool = False, 

50 check: bool = False, 

51 capture_output: bool = False, 

52 env: dict[str, str] | None = None, 

53 cwd: Optional["StrOrBytesPath"] = None, 

54 group_log: bool = False, 

55) -> subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]: ... 

56 

57 

58def exec_command( 

59 command: "_StrOrListStr", 

60 *, 

61 text: bool = False, 

62 check: bool = False, 

63 capture_output: bool = False, 

64 env: dict[str, str] | None = None, 

65 cwd: Optional["StrOrBytesPath"] = None, 

66 group_log: bool = False, 

67) -> subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]: 

68 encoding = sys.stdout.encoding if text else None 

69 if env: 

70 env = os.environ | env 

71 

72 with ( 

73 log.group(f"subprocess.run: {command}") 

74 if group_log 

75 else nullcontext(logger.info("subprocess.run: %s", command)) 

76 ): 

77 return subprocess.run( 

78 command, 

79 shell=isinstance(command, str), 

80 text=text, 

81 check=check, 

82 env=env, 

83 cwd=cwd, 

84 capture_output=capture_output, 

85 encoding=encoding, 

86 ) 

87 

88 

89@overload 

90def command_stdout( 

91 command: "_StrOrListStr", 

92 *, 

93 text: Literal[True] = True, 

94 check: bool = True, 

95 cwd: Optional["StrOrBytesPath"] = None, 

96) -> str: ... 

97 

98 

99@overload 

100def command_stdout( 

101 command: "_StrOrListStr", 

102 *, 

103 text: Literal[False], 

104 check: bool = True, 

105 cwd: Optional["StrOrBytesPath"] = None, 

106) -> bytes: ... 

107 

108 

109def command_stdout( 

110 command: "_StrOrListStr", 

111 *, 

112 text: bool = True, 

113 check: bool = True, 

114 cwd: Optional["StrOrBytesPath"] = None, 

115) -> str | bytes: 

116 return exec_command( 

117 command, 

118 text=text, 

119 cwd=cwd, 

120 check=check, 

121 capture_output=True, 

122 ).stdout