Coverage for pyTooling/CLI/Pipeline.py: 77%

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 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.Tracing.CI.GitHub 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.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 "--trace-file", dest="traceFile", metaName="[format:]file", optional=True, 

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

112 ) 

113 @LongValuedFlag( 

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

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

116 ) 

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

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

119 """ 

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

121 

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

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

124 

125 :param args: The parsed command line. 

126 """ 

127 self._PrintHeadline() 

128 

129 outputs = self._CheckOutputs(args) 

130 self.ExitOnPreviousErrors() 

131 

132 trace = self._ReadPipeline(args) 

133 self._PrintSummary(trace) 

134 self._WriteOutputs(outputs, trace) 

135 

136 self.ExitOnPreviousErrors() 

137 

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

139 """ 

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

141 

142 :param args: The parsed command line. 

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

144 """ 

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

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

147 if value is None: 

148 continue 

149 

150 try: 

151 fileFormat, file = splitFormat(value, formats) 

152 except ValueError as ex: 

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

154 for note in ex.__notes__: 

155 self.WriteErrorNote(note) 

156 continue 

157 

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

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

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

161 continue 

162 

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

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

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

166 continue 

167 

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

169 

170 return outputs 

171 

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

173 """ 

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

175 

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

177 

178 :param args: The parsed command line. 

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

180 """ 

181 return ( 

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

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

184 ) 

185 

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

187 """ 

188 Write every output the command line asked for. 

189 

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

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

192 """ 

193 for option, fileFormat, file in outputs: 

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

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

196 trace.WriteJSONFile(file) 

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

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

199 self._WriteGantt(fileFormat, file, trace) 

200 

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

202 """ 

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

204 

205 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 

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

207 

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

209 :param file: The file to write. 

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

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

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

213 """ 

214 from pyTooling.Tracing.Render.Matplotlib import MatplotlibRenderer 

215 

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

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

218 MatplotlibRenderer(layout).Write(file) 

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

220 

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

222 """ 

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

224 

225 :param args: The parsed command line. 

226 :returns: The workflow run as a trace. 

227 """ 

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

229 if repository is None: 

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

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

232 self.Exit(2) 

233 

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

235 if runID is None: 

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

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

238 self.Exit(2) 

239 elif not runID.isdigit(): 

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

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

242 self.Exit(2) 

243 

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

245 reader = WorkflowRunReader(repository, getenv(self.ENVIRONMENT_TOKEN)) 

246 

247 return reader.ReadRun(int(runID)) 

248 

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

250 """ 

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

252 

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

254 """ 

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

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

257 self.WriteNormal(f"Wall time: {trace.Duration:.0f} s")