Coverage for pyTooling/GitLab/CLI/Pipeline.py: 96%

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""" 

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

33 

34.. rubric:: Usage 

35 

36.. code-block:: bash 

37 

38 pytooling-gitlab pipeline --gitlab-project=gitlab-org/gitlab-development-kit \ 

39 --gitlab-pipeline-id=2954836704 \ 

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

41 

42.. hint:: 

43 

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

45""" 

46from argparse import Namespace 

47from os import getenv 

48from pathlib import Path 

49from typing import ClassVar, Optional as Nullable 

50 

51from pyTooling.Common import StringEnum 

52from pyTooling.Decorators import export 

53from pyTooling.MetaClasses import ExtendedType 

54from pyTooling.Attributes.ArgParse import CommandHandler, splitFormat 

55from pyTooling.Attributes.ArgParse.Flag import LongFlag 

56from pyTooling.Attributes.ArgParse.ValuedFlag import LongValuedFlag 

57from pyTooling.Tracing import Trace 

58from pyTooling.GitLab import GITLAB_API_URL 

59from pyTooling.GitLab.Tracing import PipelineReader, runnerCategory 

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.GitLab.CLI.Application`. 

87 

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

89 Reading and writing are separate steps on purpose: the trace is the intermediate every output is derived from, so 

90 a further output is a further option and not a second reader. 

91 """ 

92 ENVIRONMENT_PROJECT: ClassVar[str] = "CI_PROJECT_ID" #: Variable naming the project inside a pipeline. 

93 ENVIRONMENT_PIPELINE_ID: ClassVar[str] = "CI_PIPELINE_ID" #: Variable naming the pipeline inside a pipeline. 

94 ENVIRONMENT_API_URL: ClassVar[str] = "CI_API_V4_URL" #: Variable naming the REST API inside a pipeline. 

95 ENVIRONMENT_TOKEN: ClassVar[str] = "GITLAB_TOKEN" #: Variable holding the token to read the pipeline with. 

96 

97 @CommandHandler( 

98 "pipeline", 

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

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

101 ) 

102 @LongValuedFlag( 

103 "--gitlab-project", dest="gitlabProject", metaName="path|ID", optional=True, 

104 help=f"Project the pipeline belongs to, as its path or identifier. Default: ${ENVIRONMENT_PROJECT}." 

105 ) 

106 @LongValuedFlag( 

107 "--gitlab-pipeline-id", dest="gitlabPipelineID", metaName="ID", optional=True, 

108 help=f"Identifier of the GitLab pipeline. Default: ${ENVIRONMENT_PIPELINE_ID}." 

109 ) 

110 @LongValuedFlag( 

111 "--gitlab-api-url", dest="gitlabAPIURL", metaName="URL", optional=True, 

112 help=f"GitLab's REST API. Default: ${ENVIRONMENT_API_URL}, else {GITLAB_API_URL}." 

113 ) 

114 @LongValuedFlag( 

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

116 help=f"Token reading the pipeline, e.g. with scope 'read_api'. Default: ${ENVIRONMENT_TOKEN}." 

117 ) 

118 @LongValuedFlag( 

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

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

121 ) 

122 @LongValuedFlag( 

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

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

125 ) 

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

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

128 """ 

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

130 

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

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

133 

134 :param args: The parsed command line. 

135 """ 

136 self._PrintHeadline() 

137 

138 outputs = self._CheckOutputs(args) 

139 self.ExitOnPreviousErrors() 

140 

141 trace = self._ReadPipeline(args) 

142 self._PrintSummary(trace) 

143 self._WriteOutputs(outputs, trace) 

144 

145 self.ExitOnPreviousErrors() 

146 

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

148 """ 

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

150 

151 :param args: The parsed command line. 

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

153 """ 

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

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

156 if value is None: 

157 continue 

158 

159 try: 

160 fileFormat, file = splitFormat(value, formats) 

161 except ValueError as ex: 

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

163 for note in ex.__notes__: 

164 self.WriteErrorNote(note) 

165 continue 

166 

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

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

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

170 continue 

171 

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

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

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

175 continue 

176 

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

178 

179 return outputs 

180 

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

182 """ 

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

184 

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

186 

187 :param args: The parsed command line. 

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

189 """ 

190 return ( 

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

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

193 ) 

194 

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

196 """ 

197 Write every output the command line asked for. 

198 

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

200 :param trace: The pipeline as a trace. 

201 """ 

202 for option, fileFormat, file in outputs: 

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

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

205 trace.WriteJSONFile(file) 

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

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

208 self._WriteGantt(fileFormat, file, trace) 

209 

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

211 """ 

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

213 

214 A job is colored by its runner, as :func:`~pyTooling.GitLab.Tracing.runnerCategory` names it. 

215 

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

217 :param file: The file to write. 

218 :param trace: The pipeline as a trace. 

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

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

221 """ 

222 from pyTooling.Tracing.Render.Matplotlib import MatplotlibRenderer 

223 

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

225 layout = GanttLayout(trace, spanFilter=ciSpanFilter(), categorize=runnerCategory) 

226 MatplotlibRenderer(layout).Write(file) 

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

228 

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

230 """ 

231 Read the pipeline the command line names into a trace. 

232 

233 The token is ``--token``, else ``$GITLAB_TOKEN``; a public project's pipeline is read without one. 

234 

235 :param args: The parsed command line. 

236 :returns: The pipeline as a trace. 

237 """ 

238 project = args.gitlabProject if args.gitlabProject is not None else getenv(self.ENVIRONMENT_PROJECT) 

239 if project is None: 

240 self.WriteError("No project given.") 

241 self.WriteErrorNote(f"Set '--gitlab-project=group/project', or ${self.ENVIRONMENT_PROJECT}.") 

242 self.Exit(2) 

243 elif project.strip("/") == "" or "" in project.split("/"): 

244 self.WriteError(f"Project '{project}' is neither a path nor an identifier.") 

245 self.WriteErrorNote("E.g. '--gitlab-project=gitlab-org/gitlab-development-kit' or '--gitlab-project=74823'.") 

246 self.Exit(2) 

247 

248 pipelineID = args.gitlabPipelineID if args.gitlabPipelineID is not None else getenv(self.ENVIRONMENT_PIPELINE_ID) 

249 if pipelineID is None: 

250 self.WriteError("No pipeline given.") 

251 self.WriteErrorNote(f"Set '--gitlab-pipeline-id=<ID>', or ${self.ENVIRONMENT_PIPELINE_ID}.") 

252 self.Exit(2) 

253 elif not pipelineID.isdigit(): 

254 self.WriteError(f"Pipeline '{pipelineID}' isn't a number.") 

255 self.WriteErrorNote("It is the number in the pipeline's URL: .../-/pipelines/2954836704.") 

256 self.Exit(2) 

257 

258 if (apiURL := args.gitlabAPIURL) is None: 258 ↛ 261line 258 didn't jump to line 261 because the condition on line 258 was always true

259 apiURL = getenv(self.ENVIRONMENT_API_URL, str(GITLAB_API_URL)) 

260 

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

262 

263 self.WriteVerbose(f"Reading pipeline {pipelineID} of '{project}' from '{apiURL}' ...") 

264 reader = PipelineReader(int(project) if project.isdigit() else project, token, apiURL=apiURL) 

265 

266 return reader.ReadPipeline(int(pipelineID)) 

267 

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

269 """ 

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

271 

272 :param trace: The pipeline as a trace. 

273 """ 

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

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

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