Coverage for src/prints/histories.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from os import environ 

5from time import time 

6from typing import List, Optional 

7 

8# Components 

9from ..package.bundle import Bundle 

10from ..prints.colors import Colors 

11from ..system.notify import Notify 

12from ..system.platform import Platform 

13 

14# TimedHistory class, pylint: disable=too-few-public-methods 

15class TimedHistory: 

16 

17 # Members 

18 __duration: str 

19 __start_time: float 

20 

21 # Constructor 

22 def __init__(self) -> None: 

23 

24 # Initialize members 

25 self.__duration = '0 second' 

26 self.__start_time = time() 

27 

28 # Refresh times 

29 def _refresh_times(self) -> None: 

30 

31 # Acquire fake duration 

32 if environ.get(Bundle.ENV_HISTORIES_DURATION_FAKE, ''): 

33 duration = int(environ[Bundle.ENV_HISTORIES_DURATION_FAKE]) 

34 

35 # Evaluate duration 

36 else: 

37 duration = int(time() - self.__start_time) 

38 

39 # Evaluate seconds 

40 seconds = f"{duration % 60:.0f} second{'s' if duration % 60 > 1 else ''}" 

41 

42 # Evaluate minutes 

43 minutes = '' 

44 if duration >= 60: 

45 minutes = f"{duration / 60:.0f} minute{'s' if duration / 60 > 1 else ''} " 

46 

47 # Store total time 

48 self.__duration = minutes + seconds 

49 

50 @property 

51 def duration(self) -> str: 

52 return self.__duration 

53 

54# JobHistory class 

55class JobHistory(TimedHistory): 

56 

57 # Constants 

58 __SYMBOL_FAILED = '✘' if Platform.IS_TTY_UTF8 else 'x' 

59 __SYMBOL_FOOTER = '‣' if Platform.IS_TTY_UTF8 else '>' 

60 __SYMBOL_SKIPPED = '»' if Platform.IS_TTY_UTF8 else '~' 

61 __SYMBOL_SUCCESS = '✔' if Platform.IS_TTY_UTF8 else 'v' 

62 __SYMBOL_WARNING = '!' 

63 

64 # Members 

65 __details: str = '' 

66 __failure_allowed: bool 

67 __interrupted: bool 

68 __name: str 

69 __result: Optional[bool] 

70 __stage: str 

71 

72 # Constructor 

73 def __init__(self, name: str, stage: str) -> None: 

74 

75 # Initialize members 

76 super().__init__() 

77 self.__details = '' 

78 self.__failure_allowed = False 

79 self.__interrupted = False 

80 self.__name = name 

81 self.__result = None 

82 self.__stage = stage 

83 

84 @property 

85 def failure_allowed(self) -> bool: 

86 return self.__failure_allowed 

87 

88 @failure_allowed.setter 

89 def failure_allowed(self, value: bool) -> None: 

90 self.__failure_allowed = value 

91 

92 @property 

93 def details(self) -> str: 

94 return self.__details 

95 

96 @details.setter 

97 def details(self, value: str) -> None: 

98 self.__details = value 

99 

100 @property 

101 def interrupted(self) -> Optional[bool]: 

102 return self.__interrupted 

103 

104 @interrupted.setter 

105 def interrupted(self, value: bool) -> None: 

106 self.__interrupted = value 

107 

108 @property 

109 def name(self) -> str: 

110 return self.__name 

111 

112 @property 

113 def result(self) -> Optional[bool]: 

114 return self.__result 

115 

116 @result.setter 

117 def result(self, value: bool) -> None: 

118 self.__result = value 

119 self._refresh_times() 

120 

121 # Header 

122 def header( 

123 self, 

124 jobs_count: int, 

125 image: str, 

126 engine_type: str, 

127 ) -> None: 

128 

129 # Header output 

130 if jobs_count > 1: 

131 print(' ') 

132 print( 

133 f' {Colors.GREEN}===[ {Colors.YELLOW}{self.__stage}:' \ 

134 f' {Colors.YELLOW}{self.__name} {Colors.CYAN}' \ 

135 f'({image}, {engine_type}) {Colors.GREEN}]==={Colors.RESET}' 

136 ) 

137 print(' ') 

138 Platform.flush() 

139 

140 # Footer 

141 def footer(self) -> None: 

142 

143 # Result 

144 result: str 

145 if self.interrupted: 

146 result = 'Interrupted after' 

147 elif self.result: 

148 result = 'Success in' 

149 else: 

150 result = 'Failure in' 

151 

152 # Footer output 

153 print( 

154 f' {Colors.YELLOW}{self.__SYMBOL_FOOTER} {self.__name}:' \ 

155 f' {Colors.GREEN if self.result else Colors.RED}' \ 

156 f'{result} {self.duration}{Colors.CYAN}{self.details}{Colors.RESET}' 

157 ) 

158 print(' ') 

159 Platform.flush() 

160 

161 # Print 

162 def print(self) -> None: 

163 

164 # Variables 

165 icon = '' 

166 summary = '' 

167 

168 # Prepare job result 

169 if self.result: 

170 icon = f'{Colors.GREEN}{self.__SYMBOL_SUCCESS}' 

171 summary = f'{Colors.GREEN}Success in {self.duration}' 

172 elif self.failure_allowed: 

173 icon = f'{Colors.YELLOW}{self.__SYMBOL_WARNING}' 

174 summary = f'{Colors.YELLOW}Failure in {self.duration}' 

175 elif self.result is None: 

176 icon = f'{Colors.GREY}{self.__SYMBOL_SKIPPED}' 

177 summary = f'{Colors.GREY}Skipped' 

178 else: 

179 icon = f'{Colors.RED}{self.__SYMBOL_FAILED}' 

180 summary = f'{Colors.RED}Failure in {self.duration}' 

181 

182 # Prepare interrupted result 

183 if self.interrupted: 

184 summary += f' {Colors.YELLOW}(Interrupted)' 

185 

186 # Print result 

187 print( 

188 f' {icon} {Colors.BOLD}{self.name}:' \ 

189 f' {summary}{Colors.CYAN}{self.details}{Colors.RESET}' 

190 ) 

191 

192# StageHistory class 

193class StageHistory: 

194 

195 # Constants 

196 __SYMBOL_STAGE = '•' if Platform.IS_TTY_UTF8 else '-' 

197 

198 # Members 

199 __jobs: List[JobHistory] = [] 

200 __name: str 

201 

202 # Constructor 

203 def __init__(self, name: str) -> None: 

204 

205 # Initialize members 

206 self.__jobs = [] 

207 self.__name = name 

208 

209 @property 

210 def name(self) -> str: 

211 return self.__name 

212 

213 # Add 

214 def add(self, job_name: str) -> JobHistory: 

215 

216 # Add job 

217 job = JobHistory(job_name, self.name) 

218 self.__jobs += [job] 

219 

220 # Result 

221 return job 

222 

223 # Print 

224 def print(self) -> None: 

225 

226 # Stage header 

227 print(f' {Colors.YELLOW}{self.__SYMBOL_STAGE} Stage {self.name}:{Colors.RESET}') 

228 

229 # Iterate through jobs 

230 for job in self.__jobs: 

231 job.print() 

232 

233# PipelineHistory class 

234class PipelineHistory(TimedHistory): 

235 

236 # Constants 

237 __SYMBOL_PIPELINE = '‣' if Platform.IS_TTY_UTF8 else '>' 

238 

239 # Members 

240 __interacted: bool 

241 __jobs_count: int 

242 __jobs_quiet: bool 

243 __pipeline: List[StageHistory] 

244 __result: Optional[bool] 

245 

246 # Constructor 

247 def __init__(self) -> None: 

248 

249 # Initialize members 

250 super().__init__() 

251 self.__interacted = False 

252 self.__jobs_count = 0 

253 self.__jobs_quiet = True 

254 self.__pipeline = [] 

255 self.__result = None 

256 

257 @property 

258 def interacted(self) -> bool: 

259 return self.__interacted 

260 

261 @interacted.setter 

262 def interacted(self, value: bool) -> None: 

263 self.__interacted = value 

264 

265 @property 

266 def jobs_count(self) -> int: 

267 return self.__jobs_count 

268 

269 @property 

270 def jobs_quiet(self) -> bool: 

271 return self.__jobs_quiet 

272 

273 @jobs_quiet.setter 

274 def jobs_quiet(self, value: bool) -> None: 

275 self.__jobs_quiet = value 

276 

277 @property 

278 def result(self) -> Optional[bool]: 

279 return self.__result 

280 

281 @result.setter 

282 def result(self, value: bool) -> None: 

283 self.__result = value 

284 self._refresh_times() 

285 

286 # Add 

287 def add( 

288 self, 

289 stage_name: str, 

290 job_name: str, 

291 ) -> JobHistory: 

292 

293 # Increment jobs count 

294 self.__jobs_count += 1 

295 

296 # Find stage 

297 stage = self.get(stage_name) 

298 

299 # Prepare stage 

300 if not stage: 

301 stage = StageHistory(stage_name) 

302 self.__pipeline += [stage] 

303 

304 # Add job 

305 job = stage.add(job_name) 

306 

307 # Result 

308 return job 

309 

310 # Get 

311 def get(self, stage_name: str) -> Optional[StageHistory]: 

312 

313 # Find stage 

314 for stage in self.__pipeline: 

315 if stage.name == stage_name: 

316 return stage 

317 

318 # Fallback 

319 return None 

320 

321 # Print 

322 def print(self) -> None: 

323 

324 # Header 

325 print(' ') 

326 print( 

327 f' {Colors.GREEN}===[ {Colors.YELLOW}Pipeline:' \ 

328 f' {Colors.BOLD}{self.jobs_count} jobs {Colors.GREEN}]==={Colors.RESET}' 

329 ) 

330 print(' ') 

331 

332 # Iterate through stages 

333 for stage in self.__pipeline: 

334 stage.print() 

335 print(' ') 

336 

337 # Footer 

338 print( 

339 f' {Colors.YELLOW}{self.__SYMBOL_PIPELINE} Pipeline:' \ 

340 f' {Colors.BOLD if self.result else Colors.RED}' \ 

341 f"{'Success' if self.result else 'Failure'}" \ 

342 f' in {self.duration} total{Colors.RESET}' 

343 ) 

344 print(' ') 

345 

346 # Notify 

347 def notify(self) -> None: 

348 

349 # Notify pipeline result 

350 Notify().notify('Pipeline: ' \ 

351 f"{'Success' if self.result else 'Failure'}" \ 

352 f' in {self.duration} total')