Coverage for pyTooling/GitHub/CLI/Pipeline.py: 87%

1# ==================================================================================================================== # 

2# _____ _ _ ____ _ _ _ _ _ # 

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ / ___(_) |_| | | |_ _| |__ # 

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` || | _| | __| |_| | | | | '_ \ # 

5# | |_) | |_| || | (_) | (_) | | | | | | (_| || |_| | | |_| _ | |_| | |_) | # 

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____|_|\__|_| |_|\__,_|_.__/ # 

7# |_| |___/ |___/ # 

8# ==================================================================================================================== # 

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

13# ==================================================================================================================== # 

14# Copyright 2026-2026 Patrick Lehmann - Bötzingen, Germany # 

15# # 

16# Licensed under the Apache License, Version 2.0 (the "License"); # 

17# you may not use this file except in compliance with the License. # 

18# You may obtain a copy of the License at # 

19# # 

20# http://www.apache.org/licenses/LICENSE-2.0 # 

21# # 

22# Unless required by applicable law or agreed to in writing, software # 

23# distributed under the License is distributed on an "AS IS" BASIS, # 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

25# See the License for the specific language governing permissions and # 

26# limitations under the License. # 

27# # 

28# SPDX-License-Identifier: Apache-2.0 # 

29# ==================================================================================================================== # 

30# 

31# 

32""" 

33The :pycode:`pipeline` command: read a CI pipeline and write what it took. 

34 

35.. rubric:: Usage 

36 

37.. code-block:: bash 

38 

39 pytooling-github pipeline --github-repository=pyTooling/pyTooling \ 

40 --github-pipeline-id=35479251694 \ 

41 --trace-file=report/Pipeline.otlp.json 

42 

43.. hint:: 

44 

45 See :ref:`high-level help <CLI/Pipeline>` for explanations and usage examples. 

46""" 

47from argparse import Namespace 

48from os import getenv 

49from pathlib import Path 

50from typing import ClassVar, Optional as Nullable 

51 

52from pyTooling.Common import StringEnum 

53from pyTooling.Decorators import export 

54from pyTooling.MetaClasses import ExtendedType 

55from pyTooling.Attributes.ArgParse import CommandHandler, splitFormat 

56from pyTooling.Attributes.ArgParse.Flag import LongFlag 

57from pyTooling.Attributes.ArgParse.ValuedFlag import LongValuedFlag 

58from pyTooling.Tracing import Trace 

59from pyTooling.GitHub.Tracing import WorkflowRunReader 

60from pyTooling.Tracing.Render import GanttLayout, ciSpanFilter 

61 

62 

63@export 

64class TraceFormat(StringEnum): 

65 """The formats a trace can be written in, as ``--trace-file`` names them.""" 

66 

67 OTLPJSON = "otlp-json" #: OpenTelemetry's OTLP/JSON encoding of a trace. 

68 

69 DEFAULT = OTLPJSON #: The format ``--trace-file`` writes when its value names none. 

70 

71 

72@export 

73class GanttFormat(StringEnum): 

74 """The formats a Gantt chart can be drawn in, as ``--gantt`` names them: the backend and the file format.""" 

75 

76 MatplotlibPNG = "matplotlib-png" #: A raster image, drawn by matplotlib. 

77 MatplotlibSVG = "matplotlib-svg" #: A vector image, drawn by matplotlib. 

78 MatplotlibPDF = "matplotlib-pdf" #: A PDF page, drawn by matplotlib. 

79 

80 DEFAULT = MatplotlibPNG #: The format ``--gantt`` draws when its value names none. 

81 

82 

83@export 

84class PipelineHandlers(metaclass=ExtendedType, mixin=True): 

85 """ 

86 Mixin-class contributing the :pycode:`pipeline` command to :class:`~pyTooling.GitHub.CLI.Application`. 

87 

88 The command reads one run of a CI pipeline into a :class:`~pyTooling.Tracing.Trace` and writes what was asked 

89 of it. Reading and writing are separate steps on purpose: the trace is the intermediate every output is derived 

90 from, so a further output is a further option and not a second reader. 

91 """ 

92 ENVIRONMENT_REPOSITORY: ClassVar[str] = "GITHUB_REPOSITORY" #: Variable naming the repository inside a workflow. 

93 ENVIRONMENT_RUN_ID: ClassVar[str] = "GITHUB_RUN_ID" #: Variable naming the run inside a workflow. 

94 ENVIRONMENT_TOKEN: ClassVar[str] = "GITHUB_TOKEN" #: Variable holding the token to read the run with. 

95 

96 @CommandHandler( 

97 "pipeline", 

98 help="Read a CI pipeline run and write what it took.", 

99 description="Read a CI pipeline run and write what it took." 

100 ) 

101 @LongValuedFlag( 

102 "--github-repository", dest="githubRepository", metaName="owner/name", optional=True, 

103 help=f"Repository the workflow run belongs to. Default: ${ENVIRONMENT_REPOSITORY}." 

104 ) 

105 @LongValuedFlag( 

106 "--github-pipeline-id", dest="githubPipelineID", metaName="ID", optional=True, 

107 help=f"Identifier of the GitHub Actions workflow run. Default: ${ENVIRONMENT_RUN_ID}." 

108 ) 

109 @LongValuedFlag( 

110 "--token", dest="token", metaName="token", optional=True, 

111 help=f"Token to read the workflow run with. Default: ${ENVIRONMENT_TOKEN}." 

112 ) 

113 @LongValuedFlag( 

114 "--trace-file", dest="traceFile", metaName="[format:]file", optional=True, 

115 help=f"Write the trace. Format: {', '.join(TraceFormat)}. Default: {TraceFormat.DEFAULT}." 

116 ) 

117 @LongValuedFlag( 

118 "--gantt", dest="gantt", metaName="[format:]file", optional=True, 

119 help=f"Draw a Gantt chart. Format: {', '.join(GanttFormat)}. Default: {GanttFormat.DEFAULT}." 

120 ) 

121 @LongFlag("--force", dest="force", help="Overwrite files that exist.") 

122 def HandlePipeline(self, args: Namespace) -> None: 

123 """ 

124 Handle program calls with command ``pipeline``. 

125 

126 The outputs are checked **before** the pipeline is read, so a misspelled format or a file that exists is 

127 reported at once instead of after a network round-trip. 

128 

129 :param args: The parsed command line. 

130 """ 

131 self._PrintHeadline() 

132 

133 outputs = self._CheckOutputs(args) 

134 self.ExitOnPreviousErrors() 

135 

136 trace = self._ReadPipeline(args) 

137 self._PrintSummary(trace) 

138 self._WriteOutputs(outputs, trace) 

139 

140 self.ExitOnPreviousErrors() 

141 

142 def _CheckOutputs(self, args: Namespace) -> list[tuple[str, StringEnum, Path]]: 

143 """ 

144 Read every output option, and report what can't be written before anything is read. 

145 

146 :param args: The parsed command line. 

147 :returns: One ``(option, format, file)`` per output that was asked for and can be written. 

148 """ 

149 outputs: list[tuple[str, StringEnum, Path]] = [] 

150 for option, value, formats in self._Outputs(args): 

151 if value is None: 

152 continue 

153 

154 try: 

155 fileFormat, file = splitFormat(value, formats) 

156 except ValueError as ex: 

157 self.WriteError(f"Option '{option}': {ex}") 

158 for note in ex.__notes__: 

159 self.WriteErrorNote(note) 

160 continue 

161 

162 if option == "--gantt" and file.suffix.lower().lstrip(".") != (suffix := fileFormat.partition("-")[2]): 

163 self.WriteError(f"Option '--gantt': format '{fileFormat}' writes a '.{suffix}' file.") 

164 self.WriteErrorNote(f"Got '{file.name}'. Name the file '.{suffix}', or state the format it is in.") 

165 continue 

166 

167 if file.exists() and not args.force: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true

168 self.WriteError(f"File '{file}' exists.") 

169 self.WriteErrorNote("Use '--force' to overwrite it.") 

170 continue 

171 

172 outputs.append((option, fileFormat, file)) 

173 

174 return outputs 

175 

176 def _Outputs(self, args: Namespace) -> tuple[tuple[str, Nullable[str], type[StringEnum]], ...]: 

177 """ 

178 Return the output options this command offers, as ``(option, value, formats)``. 

179 

180 A value naming no format gets the enumeration's ``DEFAULT``. 

181 

182 :param args: The parsed command line. 

183 :returns: One entry per output option, whether or not it was given. 

184 """ 

185 return ( 

186 ("--trace-file", args.traceFile, TraceFormat), 

187 ("--gantt", args.gantt, GanttFormat), 

188 ) 

189 

190 def _WriteOutputs(self, outputs: list[tuple[str, StringEnum, Path]], trace: Trace) -> None: 

191 """ 

192 Write every output the command line asked for. 

193 

194 :param outputs: The outputs, as :meth:`_CheckOutputs` returned them. 

195 :param trace: The workflow run as a trace. 

196 """ 

197 for option, fileFormat, file in outputs: 

198 if option == "--trace-file": 

199 self.WriteVerbose(f"Writing the trace as '{fileFormat}' to '{file}' ...") 

200 trace.WriteJSONFile(file) 

201 self.WriteNormal(f"Trace: {file}") 

202 elif option == "--gantt": 202 ↛ 197line 202 didn't jump to line 197 because the condition on line 202 was always true

203 self._WriteGantt(fileFormat, file, trace) 

204 

205 def _WriteGantt(self, fileFormat: GanttFormat, file: Path, trace: Trace) -> None: 

206 """ 

207 Lay the trace out as a Gantt chart and draw it with the backend the format names. 

208 

209 The steps of a job are left out: a pipeline of 57 jobs has more than a thousand steps, and a chart of one row 

210 per step is a different picture than a chart of one row per job. 

211 

212 :param fileFormat: The format, one of :class:`GanttFormat`. 

213 :param file: The file to write. 

214 :param trace: The workflow run as a trace. 

215 :raises MissingDependencyError: If *matplotlib* isn't installed. |br| 

216 :func:`~pyTooling.GitHub.CLI.main` prints it with the commands installing it. 

217 """ 

218 from pyTooling.Tracing.Render.Matplotlib import MatplotlibRenderer 

219 

220 self.WriteVerbose(f"Drawing the Gantt chart as '{fileFormat}' to '{file}' ...") 

221 layout = GanttLayout(trace, spanFilter=ciSpanFilter()) 

222 MatplotlibRenderer(layout).Write(file) 

223 self.WriteNormal(f"Gantt: {file} ({layout.RowCount} rows)") 

224 

225 def _ReadPipeline(self, args: Namespace) -> Trace: 

226 """ 

227 Read the workflow run the command line names into a trace. 

228 

229 :param args: The parsed command line. 

230 :returns: The workflow run as a trace. 

231 """ 

232 repository = args.githubRepository if args.githubRepository is not None else getenv(self.ENVIRONMENT_REPOSITORY) 

233 if repository is None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 self.WriteError("No repository given.") 

235 self.WriteErrorNote(f"Set '--github-repository=owner/name', or ${self.ENVIRONMENT_REPOSITORY}.") 

236 self.Exit(2) 

237 

238 owner, _, name = repository.partition("/") 

239 if owner == "" or name == "" or "/" in name: 

240 self.WriteError(f"Repository '{repository}' isn't of the form 'owner/name'.") 

241 self.WriteErrorNote("E.g. '--github-repository=pyTooling/pyTooling'.") 

242 self.Exit(2) 

243 

244 runID = args.githubPipelineID if args.githubPipelineID is not None else getenv(self.ENVIRONMENT_RUN_ID) 

245 if runID is None: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true

246 self.WriteError("No workflow run given.") 

247 self.WriteErrorNote(f"Set '--github-pipeline-id=<ID>', or ${self.ENVIRONMENT_RUN_ID}.") 

248 self.Exit(2) 

249 elif not runID.isdigit(): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 self.WriteError(f"Workflow run '{runID}' isn't a number.") 

251 self.WriteErrorNote("It is the number in the run's URL: .../actions/runs/35479251694.") 

252 self.Exit(2) 

253 

254 self.WriteVerbose(f"Reading run {runID} of '{repository}' ...") 

255 token = args.token if args.token is not None else getenv(self.ENVIRONMENT_TOKEN) 

256 reader = WorkflowRunReader(owner, name, token) 

257 

258 return reader.ReadRun(int(runID)) 

259 

260 def _PrintSummary(self, trace: Trace) -> None: 

261 """ 

262 Print what the pipeline took, so the command says something without being asked to write a file. 

263 

264 :param trace: The workflow run as a trace. 

265 """ 

266 self.WriteNormal(f"Pipeline: {trace.Name}") 

267 self.WriteNormal(f"Began: {trace.StartTime}") 

268 self.WriteNormal(f"Wall time: {trace.DurationInSeconds:.0f} s")