Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/invocation.py: 81%

1import logging 

2import pathlib 

3import subprocess 

4from dataclasses import dataclass, field 

5from datetime import timedelta 

6from timeit import default_timer as timer 

7 

8from .exceptions import CoxbuildRuntimeException 

9 

10logger = logging.getLogger("invocation") 

11 

12 

13class CommandExecutionException(CoxbuildRuntimeException): 

14 """Exception for command execution.""" 

15 

16 def __init__(self, result: "CommandExecutionResult", error: str = ""): 

17 message = f"Command {' '.join(result.args.cmds)} failed with error {error}" 

18 if result.timeout: 

19 message += " (timeout)" 

20 else: 

21 message += f" (exit code {result.code})" 

22 if result.stdout: 

23 message += f"\nstdout: {result.stdout}" 

24 if result.stderr: 

25 message += f"\nstderr: {result.stderr}" 

26 super().__init__(message) 

27 self.result = result 

28 self.error = error 

29 

30 

31@dataclass 

32class CommandExecutionResult: 

33 """Result for command execution.""" 

34 args: "CommandExecutionArgs" 

35 """arguments for command execution""" 

36 duration: timedelta = field(default_factory=timedelta) 

37 """execution duration""" 

38 code: int | None = None 

39 """exit code, None for timeout""" 

40 stdout: str = "" 

41 """stdout in text (if pipe)""" 

42 stderr: str = "" 

43 """stderr in text (if pipe)""" 

44 

45 @property 

46 def timeout(self) -> bool: 

47 """Return if execution timeout.""" 

48 return self.code is None 

49 

50 def __bool__(self): 

51 return self.code == 0 

52 

53 @property 

54 def description(self): 

55 """Return result's description string.""" 

56 return "🟢 SUCCESS" if self else ("🟡 TIMEOUT" if self.timeout else f"🔴 FAILING({self.code})") 

57 

58 def ensure(self): 

59 if not self: 

60 raise CommandExecutionException(self) 

61 

62 

63def execmd(args: "CommandExecutionArgs"): 

64 """Execute command and get result.""" 

65 logger.debug(f"Execute command: {args}") 

66 

67 result = CommandExecutionResult(args) 

68 

69 tic = timer() 

70 try: 

71 subresult = subprocess.run(args=args.cmds, env=args.env, cwd=args.cwd, encoding="utf-8", text=True, input=args.input, shell=args.shell, 

72 timeout=args.timeout.total_seconds() if args.timeout else None, capture_output=args.pipe) 

73 result.code = subresult.returncode 

74 result.stdout = subresult.stdout or "" 

75 result.stderr = subresult.stderr or "" 

76 except subprocess.TimeoutExpired as te: 

77 result.stdout = te.stdout or "" 

78 result.stderr = te.stderr or "" 

79 

80 result.duration = timedelta(seconds=timer()-tic) 

81 logger.info(f"Executed command: {args} -> {result}") 

82 return result 

83 

84 

85@dataclass 

86class CommandExecutionArgs: 

87 """Arguments for command execution.""" 

88 cmds: list[str] 

89 """command and arguments""" 

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

91 """environ""" 

92 cwd: pathlib.Path | None = None 

93 """current working directory""" 

94 timeout: timedelta | None = None 

95 """maximum execution duration""" 

96 input: str | None = None 

97 """text for stdin""" 

98 shell: bool = False 

99 """use system shell""" 

100 pipe: bool = False 

101 """pipe and collect stdout and stderr""" 

102 

103 def run(self, retry: int = 0, fail: bool = False) -> CommandExecutionResult: 

104 """ 

105 Run command. 

106 

107 retry: the number of times to retry when failing 

108 fail: do not raise exception when the final result fails 

109 """ 

110 

111 result = execmd(self) 

112 if not result: 

113 for i in range(retry): 

114 logger.info(f"Retry ({i+1}/{retry}) execute command: {self}") 

115 result = execmd(self) 

116 if result: 

117 break 

118 if not fail: 

119 result.ensure() 

120 return result 

121 

122 

123def run(cmds: list[str], env: dict[str, str] | None = None, 

124 cwd: pathlib.Path | None = None, timeout: timedelta | None = None, 

125 input: str | None = None, 

126 shell: bool = False, pipe: bool = False, 

127 retry: int = 0, fail: bool = False) -> CommandExecutionResult: 

128 """ 

129 Run command. 

130 

131 cmds: command and argument 

132 env: environ 

133 cwd: current working directory 

134 timeout: maximum execution duration 

135 input: text for stdin 

136 shell: use system shell 

137 pipe: pipe and collect stdout and stderr 

138 retry: the number of times to retry when failing 

139 fail: do not raise exception when the final result fails 

140 """ 

141 return CommandExecutionArgs(cmds, env, cwd, timeout, input, shell, pipe).run(retry=retry, fail=fail)