Coverage for pyTooling/GitHub/Sphinx/Graph.py: 97%

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

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

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

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

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

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

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

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

9# Authors: #

10# Patrick Lehmann #

11# #

12# License: #

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

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

32A Sphinx directive drawing the pipeline of a GitHub Actions workflow as a Graphviz graph.

34The jobs of a workflow and the ``needs`` between them are its pipeline. This directive draws it at build time from the

35workflow file:

37.. code-block:: ReST

39 .. ghactions:pipeline-graph:: ../.github/workflows/CompletePipeline.yml

40 :depth: 1

42A job is a node labelled with its name and, if it calls a reusable workflow, that workflow's file. A reusable workflow

43of the documented repository is expanded into a cluster of its own jobs, as many levels deep as ``:depth:`` says. In

44HTML, a node links to the page documenting its reusable workflow, if the ``ghactions`` domain knows one.

46.. seealso::

48 :mod:`pyTooling.GitHub.WorkflowFile`

49 |rarr| The model of a workflow file the graph is drawn from.

50 :mod:`pyTooling.CI`

51 |rarr| The service-independent model of a pipeline, which a workflow file is converted into.

52 :mod:`pyTooling.Sphinx.SchemaGraph`

53 |rarr| The other graph directives of the extension.

54"""

55from __future__ import annotations

57from html import escape as html_escape

58from pathlib import Path, PurePosixPath

59from re import Match, compile as re_compile

60from typing import TYPE_CHECKING, Any, Optional as Nullable

62from docutils import nodes

63from docutils.parsers.rst import directives

64from sphinx.application import Sphinx

65from sphinx.ext.graphviz import align_spec, figure_wrapper, graphviz

66from sphinx.util.logging import getLogger

68from pyTooling.CI import Base as CIBase, Matrix as CIMatrix, Workflow as CIWorkflow

69from pyTooling.Common import getFullyQualifiedName

70from pyTooling.Decorators import export, readonly

71from pyTooling.Graph import Vertex

72from pyTooling.MetaClasses import ExtendedType

73from pyTooling.Sphinx import BaseDirective, SphinxExtensionError, strip, stripAndNormalize

75if TYPE_CHECKING: # pragma: no cover

76 from pyTooling.GitHub.WorkflowFile import Job, Workflow, WorkflowResolver

79__all__ = ["GRAPH_ATTRIBUTES", "CSS_CLASS", "LINK_MARKER"]

81#: Attributes every pipeline graph is drawn with.

82GRAPH_ATTRIBUTES = (

83 "compound=true;",

84 "newrank=true;",

85 'fontname="sans-serif";',

86 'fontsize="10";',

87 "nodesep=0.25;",

88 "ranksep=0.4;",

89 'node [shape="box", style="rounded,filled", color="#5f6b7a", fillcolor="#e4ecf7", penwidth="1.0",',

90 ' fontname="sans-serif", fontsize="10"];',

91 'edge [color="#5f6b7a", arrowsize="0.7"];',

92)

94#: CSS class of a pipeline graph's ``graphviz`` node, which is also how :func:`resolveLinks` finds it.

95CSS_CLASS = "ghactions-pipeline-graph"

97#: A DOT comment, and the whitespace before it, standing where a job's link belongs, until :func:`resolveLinks`

98#: replaces it by the link or removes it.

99LINK_MARKER = re_compile(r"(?P<space>\s*)/\*ghactions-link:(?P<stem>[^*]*)\*/")

100

101_logger = getLogger(__name__)

102

103

104@export

105class PipelineDotGraph(metaclass=ExtendedType, slots=True):

106 """

107 The pipeline of a workflow in the DOT language: its jobs as nodes, their ``needs`` as edges.

108

109 The workflow is converted by :meth:`Workflow.ToPipeline <pyTooling.GitHub.WorkflowFile.Workflow.ToPipeline>` into a

110 :mod:`pyTooling.CI` model, and that by :meth:`~pyTooling.CI.Workflow.ToGraph` into a

111 :class:`~pyTooling.Graph.Graph`, which is drawn: a vertex is a node, an edge an edge, and a called workflow with a

112 :class:`~pyTooling.Graph.Subgraph` a cluster of the vertices it links to.

113

114 The kind of an element is its node's shape and style:

115

116 * a job calling a reusable workflow is a rounded box, labelled with the job's name and the workflow's file;

117 * a job calling a reusable workflow of another repository is a white rounded box, and its label also names that

118 repository and ref - it is never expanded;

119 * a job running steps is a grey box with square corners;

120 * a job with an ``if`` condition is dashed, and its tooltip is the condition;

121 * a job with a ``strategy.matrix`` is a cluster of its instances, and its label names the matrix' dimensions: an

122 instance of a job running steps is a job's node, an instance calling a reusable workflow is drawn as a job calling

123 it. A dynamic matrix - its combinations known only at run time - is one node with a double border.

124

125 A job calling a reusable workflow that :class:`~pyTooling.GitHub.WorkflowFile.WorkflowResolver` finds locally is

126 drawn as a cluster of that workflow's jobs instead, until the depth is used up - an instance of a matrix as well. An

127 edge to or from a job drawn as a cluster ends at the cluster's border.

128

129 The graph's edges read *needs*; an arrow is drawn the other way round, from the needed job to the job needing it,

130 in the direction the pipeline runs.

131

132 The graph is drawn in file order, so the same workflow is always drawn as the same DOT.

133 """

134

135 _resolver: WorkflowResolver #: Resolver reading the reusable workflows the jobs call.

136 _link: bool #: Whether a job calling a reusable workflow of the documented repository is linked.

137 _statements: list[str] #: The statements written so far, one per line, already indented.

138 _jobs: list[Job] #: The jobs drawn, each once, in the order they were drawn.

139 _workflows: list[Workflow] #: The workflows drawn, each once, the entry-point workflow first.

140

141 def __init__(

142 self,

143 workflow: Workflow,

144 resolver: Nullable[WorkflowResolver] = None,

145 direction: str = "LR",

146 depth: int = 0,

147 reduce: bool = True,

148 link: bool = True

149 ) -> None:

150 """

151 Initializes the graph by drawing a workflow.

152

153 :param workflow: The workflow to draw.

154 :param resolver: Optional, the resolver reading the reusable workflows the jobs call. It also decides which

155 repository is documented: a local one, and every repository it maps to a directory.

156 Default: a resolver knowing local references only.

157 :param direction: Optional, direction the pipeline flows in: ``LR`` (left to right) or ``TB`` (top to

158 bottom). Default: ``LR``.

159 :param depth: Optional, levels of reusable workflows expanded into clusters. Default: ``0``.

160 :param reduce: Optional, drop an edge a longer path implies. Default: ``True``.

161 :param link: Optional, mark a job calling a reusable workflow of the documented repository for

162 :func:`resolveLinks`. Default: ``True``.

163 :raises ValueError: If parameter 'workflow' is None.

164 :raises TypeError: If parameter 'workflow' is not of type :class:`~pyTooling.GitHub.WorkflowFile.Workflow`.

165 :raises TypeError: If parameter 'resolver' is not of type

166 :class:`~pyTooling.GitHub.WorkflowFile.WorkflowResolver`.

167 :raises ValueError: If parameter 'direction' is None.

168 :raises TypeError: If parameter 'direction' is not of type :class:`str`.

169 :raises ValueError: If parameter 'direction' is neither ``LR`` nor ``TB``.

170 :raises ValueError: If parameter 'depth' is None.

171 :raises TypeError: If parameter 'depth' is not of type :class:`int`.

172 :raises ValueError: If parameter 'depth' is negative.

173 :raises ValueError: If parameter 'reduce' is None.

174 :raises TypeError: If parameter 'reduce' is not of type :class:`bool`.

175 :raises ValueError: If parameter 'link' is None.

176 :raises TypeError: If parameter 'link' is not of type :class:`bool`.

177 :raises WorkflowError: If a reusable workflow to expand doesn't exist, or is not a well-formed workflow.

178 """

179 from pyTooling.GitHub.WorkflowFile import Workflow, WorkflowResolver

180

181 if workflow is None:

182 raise ValueError("Parameter 'workflow' is None.")

183 elif not isinstance(workflow, Workflow):

184 ex = TypeError("Parameter 'workflow' is not of type 'Workflow'.")

185 ex.add_note(f"Got type '{getFullyQualifiedName(workflow)}'.")

186 raise ex

187

188 if resolver is None:

189 resolver = WorkflowResolver()

190 elif not isinstance(resolver, WorkflowResolver):

191 ex = TypeError("Parameter 'resolver' is not of type 'WorkflowResolver'.")

192 ex.add_note(f"Got type '{getFullyQualifiedName(resolver)}'.")

193 raise ex

194

195 if direction is None:

196 raise ValueError("Parameter 'direction' is None.")

197 elif not isinstance(direction, str):

198 ex = TypeError("Parameter 'direction' is not of type 'str'.")

199 ex.add_note(f"Got type '{getFullyQualifiedName(direction)}'.")

200 raise ex

201 elif direction not in ("LR", "TB"):

202 ex = ValueError("Parameter 'direction' is neither 'LR' nor 'TB'.")

203 ex.add_note(f"Got value '{direction}'.")

204 raise ex

205

206 if depth is None:

207 raise ValueError("Parameter 'depth' is None.")

208 elif not isinstance(depth, int) or isinstance(depth, bool):

209 ex = TypeError("Parameter 'depth' is not of type 'int'.")

210 ex.add_note(f"Got type '{getFullyQualifiedName(depth)}'.")

211 raise ex

212 elif depth < 0:

213 ex = ValueError("Parameter 'depth' is negative.")

214 ex.add_note(f"Got value '{depth}'.")

215 raise ex

216

217 if reduce is None:

218 raise ValueError("Parameter 'reduce' is None.")

219 elif not isinstance(reduce, bool):

220 ex = TypeError("Parameter 'reduce' is not of type 'bool'.")

221 ex.add_note(f"Got type '{getFullyQualifiedName(reduce)}'.")

222 raise ex

223

224 if link is None:

225 raise ValueError("Parameter 'link' is None.")

226 elif not isinstance(link, bool):

227 ex = TypeError("Parameter 'link' is not of type 'bool'.")

228 ex.add_note(f"Got type '{getFullyQualifiedName(link)}'.")

229 raise ex

230

231 self._resolver = resolver

232 self._link = link

233 self._statements = [f"\trankdir={direction};", *(f"\t{attribute}" for attribute in GRAPH_ATTRIBUTES), ""]

234 self._jobs = []

235 self._workflows = [workflow]

236

237 graph = workflow.ToPipeline(resolver, depth).ToGraph(reduce=reduce)

238 self._DrawVertices(list(graph.IterateVertices()), "", "\t")

239

240 @readonly

241 def Jobs(self) -> list[Job]:

242 """

243 Read-only property to access the jobs drawn (:attr:`_jobs`).

244

245 A reusable workflow expanded twice is drawn twice, but its jobs are listed once.

246

247 :returns: The jobs of every workflow drawn, in the order they were drawn.

248 """

249 return self._jobs

250

251 @readonly

252 def Workflows(self) -> list[Workflow]:

253 """

254 Read-only property to access the workflows drawn (:attr:`_workflows`).

255

256 :returns: The entry-point workflow, followed by every reusable workflow expanded into a cluster, each once.

257 """

258 return self._workflows

259

260 @staticmethod

261 def _Quote(text: str) -> str:

262 """

263 Quote a text as a DOT string, keeping its line breaks as escape sequences.

264

265 :param text: The text to quote.

266 :returns: The text in double quotes.

267 """

268 text = text.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")

269

270 return f'"{text}"'

271

272 @staticmethod

273 def _Label(title: str, lines: list[str]) -> str:

274 """

275 Render an HTML-like label: a title and, below it, lines in a smaller font.

276

277 :param title: The title, unescaped.

278 :param lines: The lines below the title, unescaped.

279 :returns: The label in angle brackets, as DOT writes an HTML-like label.

280 """

281 label = html_escape(title, quote=False)

282 for line in lines:

283 label += f'<br/><font point-size="8" color="#3d4652">{html_escape(line, quote=False)}</font>'

284

285 return f"<{label}>"

286

287 def _Describe(self, element: CIBase) -> tuple[list[str], list[str], str]:

288 """

289 Describe an element of the pipeline: the lines of its label, the parts of its style, and its tooltip.

290

291 :param element: The job, matrix or called workflow to describe.

292 :returns: The label's lines below the element's name, the style's parts, and the tooltip - the reusable

293 workflow called and the condition of each rule, one per line, or an empty string.

294 """

295 lines = []

296 style = []

297 tooltip = []

298

299 if (uses := element.Definition.Uses) is not None:

300 lines.append(uses.FileName)

301 if not self._resolver.CanResolve(uses):

302 lines.append(f"{uses.Repository}@{uses.Reference}")

303 tooltip.append(f"uses: {uses}")

304

305 if isinstance(element, CIMatrix):

306 if (matrix := element.Definition.Matrix).IsDynamic or len(matrix.Dimensions) == 0:

307 lines.append("matrix")

308 else:

309 lines.append(f"matrix: {', '.join(matrix.Dimensions)}")

310

311 if element.RuleCount > 0:

312 style.append("dashed")

313 for rule in element.Rules:

314 tooltip.append(f"if: {rule.Condition}")

315

316 return lines, style, "\n".join(tooltip)

317

318 def _DrawNode(self, element: CIBase, identifier: str, indent: str) -> None:

319 """

320 Draw a job, a matrix or a called workflow that isn't expanded as a node.

321

322 :param element: The element to draw.

323 :param identifier: Identifier of the node.

324 :param indent: Indentation of the node's statement.

325 """

326 lines, style, tooltip = self._Describe(element)

327

328 if (uses := element.Definition.Uses) is None:

329 attributes = {"style": ",".join(("filled", *style)), "fillcolor": "#f2f2f2"}

330 elif self._resolver.CanResolve(uses):

331 attributes = {"style": ",".join(("rounded", "filled", *style))}

332 else:

333 attributes = {"style": ",".join(("rounded", "filled", *style)), "fillcolor": "#ffffff"}

334

335 if isinstance(element, CIMatrix):

336 attributes["peripheries"] = "2"

337

338 statement = f"{indent}{self._Quote(identifier)} [label={self._Label(str(element), lines)}"

339 for name, value in attributes.items():

340 statement += f", {name}={self._Quote(value)}"

341

342 if tooltip != "":

343 statement += f", tooltip={self._Quote(tooltip)}"

344

345 if self._link and uses is not None and self._resolver.CanResolve(uses):

346 statement += f" /*ghactions-link:{uses.Stem}*/"

347

348 self._statements.append(f"{statement}];")

349

350 def _OpenCluster(self, element: CIBase, cluster: str, indent: str) -> None:

351 """

352 Open a cluster drawing a called workflow or a matrix: write its ``subgraph`` statement and its attributes.

353

354 :param element: The called workflow or the matrix.

355 :param cluster: Identifier of the cluster.

356 :param indent: Indentation of the cluster's statement.

357 """

358 lines, style, tooltip = self._Describe(element)

359 self._statements.append(f"{indent}subgraph {self._Quote(cluster)} {{")

360 self._statements.append(f"{indent}\tlabel={self._Label(str(element), lines)};")

361 self._statements.append(f'{indent}\tlabeljust="l";')

362 self._statements.append(f"{indent}\tstyle={self._Quote(','.join(('rounded', 'filled', *style)))};")

363 self._statements.append(f'{indent}\tcolor="#8a9bb2";')

364 self._statements.append(f'{indent}\tfillcolor="#f6f8fb";')

365 if tooltip != "":

366 self._statements.append(f"{indent}\ttooltip={self._Quote(tooltip)};")

367

368 def _DrawVertices(self, vertices: list[Vertex], prefix: str, indent: str) -> tuple[str, str]:

369 """

370 Draw the vertices of one level of the graph - the graph itself, or the subgraph of a called workflow or of a

371 matrix - and their edges.

372

373 An edge of the graph goes from the element needing to the element it needs; its arrow is drawn the other way

374 round.

375

376 :param vertices: The vertices, in file order.

377 :param prefix: Prefix of the identifiers of their nodes: the path of the jobs calling their workflow.

378 :param indent: Indentation of their statements.

379 :returns: Identifiers of the node an arrow into the level ends at - the first vertex needing nothing - and

380 of the node an arrow out of it starts at - the last vertex nothing needs.

381 """

382 # per vertex: the node an arrow into it ends at, the node an arrow out of it starts at, and its cluster

383 anchors: dict[Vertex, tuple[str, str, Nullable[str]]] = {}

384 for vertex in vertices:

385 element = vertex.Value

386 if element.Definition not in self._jobs:

387 self._jobs.append(element.Definition)

388

389 identifier = f"{prefix}{element}"

390 if not isinstance(element, (CIWorkflow, CIMatrix)) or vertex.OutboundLinkCount == 0:

391 self._DrawNode(element, identifier, indent)

392 anchors[vertex] = (identifier, identifier, None)

393 continue

394

395 cluster = f"cluster_{identifier}"

396 self._OpenCluster(element, cluster, indent)

397 if isinstance(element, CIWorkflow):

398 if element.CalledWorkflow not in self._workflows:

399 self._workflows.append(element.CalledWorkflow)

400

401 if self._link: 401 ↛ 404line 401 didn't jump to line 404 because the condition on line 401 was always true

402 self._statements.append(f"{indent}\t/*ghactions-link:{element.Definition.Uses.Stem}*/")

403

404 entryNode, exitNode = self._DrawVertices(

405 [link.Destination for link in vertex.OutboundLinks], f"{identifier}/", f"{indent}\t"

406 )

407 self._statements.append(f"{indent}}}")

408 anchors[vertex] = (entryNode, exitNode, cluster)

409

410 # the vertices needing nothing start together: in the first rank of the graph, or side by side in their cluster

411 roots = [self._Quote(anchors[vertex][0]) for vertex in vertices if vertex.OutboundEdgeCount == 0]

412 if prefix == "":

413 self._statements.append(f"{indent}{{rank=min; {'; '.join(roots)};}}")

414 elif len(roots) > 1:

415 self._statements.append(f"{indent}{{rank=same; {'; '.join(roots)};}}")

416

417 for vertex in vertices:

418 for edge in vertex.OutboundEdges:

419 _, tail, tailCluster = anchors[edge.Destination]

420 head, _, headCluster = anchors[vertex]

421

422 attributes = []

423 if tailCluster is not None:

424 attributes.append(f"ltail={self._Quote(tailCluster)}")

425

426 if headCluster is not None:

427 attributes.append(f"lhead={self._Quote(headCluster)}")

428

429 statement = f"{indent}{self._Quote(tail)} -> {self._Quote(head)}"

430 if len(attributes) > 0:

431 statement += f" [{', '.join(attributes)}]"

432

433 self._statements.append(f"{statement};")

434

435 first = next(vertex for vertex in vertices if vertex.OutboundEdgeCount == 0)

436 last = [vertex for vertex in vertices if vertex.InboundEdgeCount == 0][-1]

437

438 return anchors[first][0], anchors[last][1]

439

440 def __str__(self) -> str:

441 """

442 Render the graph.

443

444 :returns: The graph in the DOT language.

445 """

446 statements = "\n".join(self._statements)

447

448 return f"digraph pipeline {{\n{statements}\n}}"

449

450

451@export

452class PipelineGraph(BaseDirective):

453 """

454 The ``ghactions:pipeline-graph`` directive: the pipeline of a workflow, drawn from the workflow file.

455

456 One argument, the path of the workflow file relative to the document using the directive.

457

458 The documented repository and the directory of its workflow files are the configuration values

459 ``ghactions_repository`` and ``ghactions_workflow_directory`` - relative to the source directory, and by default the

460 directory of the drawn workflow file. A job calling a reusable workflow of that repository is expanded, while

461 ``:depth:`` allows, and checked against the ref ``ghactions_ref``, if that is configured.

462

463 The workflow files are read by the resolver of the ``ghactions`` domain, which reads every file once per build.

464 Without ``ghactions_workflow_directory``, a resolver of its own maps ``ghactions_repository`` to the drawn file's

465 directory.

466 """

467

468 directiveName: str = "ghactions:pipeline-graph" #: Name the directive is invoked by.

469

470 has_content = False #: A boolean; ``True`` if content is allowed.

471 required_arguments = 1 #: Number of required directive arguments: the workflow file's path.

472 optional_arguments = 0 #: Number of optional arguments after the required ones.

473 final_argument_whitespace = False #: A boolean; ``True`` if the last argument may contain spaces.

474 # docutils declares 'option_spec' on 'Directive' and 'BaseDirective' assigns it, so mypy calls every

475 # spelling of this override a conflict with one of them

476 #: Mapping of option names to validator functions.

477 option_spec: dict[str, Any] = { # type: ignore[misc]

478 "align": align_spec,

479 "alt": strip,

480 "caption": strip,

481 "depth": directives.nonnegative_int,

482 "direction": strip,

483 "link": stripAndNormalize,

484 "name": strip,

485 "reduce": stripAndNormalize,

486 }

487

488 def run(self) -> list[nodes.Node]:

489 """

490 Read the workflow file and hand its pipeline to :mod:`sphinx.ext.graphviz` for rendering.

491

492 Every workflow file drawn becomes a dependency of the document. A job calling a reusable workflow of the

493 documented repository at another ref than ``ghactions_ref`` is reported as a warning of type ``ghactions.ref``.

494

495 :returns: A ``graphviz`` node, wrapped in a figure when a caption was given, or an error node when the options or

496 the workflow files are wrong.

497 """

498 relativePath, absolutePath = self.env.relfn2path(self.arguments[0])

499 self.env.note_dependency(relativePath)

500 workflowFile = Path(absolutePath)

501

502 from pyTooling.GitHub.WorkflowFile import WorkflowError, WorkflowResolver

503

504 repository = self.config.ghactions_repository

505 ref = self.config.ghactions_ref

506

507 try:

508 direction = self._ParseStringOption("direction", "LR", "(?i)(LR|TB)$").upper()

509 reduce = self._ParseBooleanOption("reduce", True)

510 link = self._ParseBooleanOption("link", True)

511 except SphinxExtensionError as ex:

512 return [self.state.document.reporter.error(str(ex), line=self.lineno)]

513

514 try:

515 if repository is not None and self.config.ghactions_workflow_directory is None:

516 resolver = WorkflowResolver({repository: workflowFile.parent})

517 else:

518 resolver = self.env.get_domain("ghactions").Resolver

519

520 graph = PipelineDotGraph(

521 resolver.Load(workflowFile), resolver, direction, self.options.get("depth", 0), reduce, link

522 )

523 except (OSError, ValueError, WorkflowError) as ex:

524 message = f"{self.directiveName}: Couldn't draw '{self.arguments[0]}': {ex}"

525 for note in getattr(ex, "__notes__", []):

526 message += f" {note}"

527

528 return [self.state.document.reporter.error(message, line=self.lineno)]

529

530 for workflow in graph.Workflows[1:]:

531 self.env.note_dependency(str(workflow.Path))

532

533 if repository is not None and ref is not None:

534 for job in graph.Jobs:

535 if (

536 (uses := job.Uses) is not None and uses.Repository is not None and

537 uses.Repository.lower() == repository.lower() and uses.Reference != ref

538 ):

539 _logger.warning(

540 f"{self.directiveName}: Job '{job.Name}' calls '{uses.FileName}' at ref '{uses.Reference}', but the "

541 f"documentation describes ref '{ref}' ({uses.Location}).",

542 location=(self.env.docname, self.lineno),

543 type="ghactions",

544 subtype="ref"

545 )

546

547 node = graphviz()

548 node["code"] = str(graph)

549 node["options"] = {"docname": self.env.docname}

550 node["alt"] = self.options.get("alt", f"Pipeline of {workflowFile.name}")

551 node["classes"] = [CSS_CLASS]

552 if "align" in self.options: 552 ↛ 553line 552 didn't jump to line 553 because the condition on line 552 was never true

553 node["align"] = self.options["align"]

554

555 if (caption := self.options.get("caption", None)) is not None: 555 ↛ 556line 555 didn't jump to line 556 because the condition on line 555 was never true

556 figure = figure_wrapper(self, node, caption)

557 self.add_name(figure)

558 return [figure]

559

560 self.add_name(node)

561 return [node]

562

563

564@export

565def resolveLinks(sphinx: Sphinx, doctree: nodes.document, docname: str) -> None:

566 """

567 Call-back for Sphinx' ``doctree-resolved`` event, linking the jobs of a pipeline graph to their reusable workflows.

568

569 A job calling a reusable workflow of the documented repository links to the page documenting that workflow, as the

570 ``ghactions`` domain resolves the workflow's file stem with ``ResolveWorkflow``. Without a page for the workflow, or

571 in a format other than HTML, the job has no link.

572

573 :param sphinx: The Sphinx application.

574 :param doctree: The resolved document.

575 :param docname: Name of the document.

576 """

577 if sphinx.builder.format != "html":

578 for node in doctree.findall(graphviz):

579 if CSS_CLASS in node["classes"]: 579 ↛ 578line 579 didn't jump to line 578 because the condition on line 579 was always true

580 node["code"] = LINK_MARKER.sub("", node["code"])

581

582 return

583

584 domain = sphinx.env.get_domain("ghactions")

585

586 def link(match: Match[str]) -> str:

587 """

588 Nested function returning the link attributes replacing a marker, or nothing.

589

590 :param match: The marker.

591 :returns: The whitespace before the marker and the attributes ``URL`` and ``target``, or an empty string

592 when there is nothing to link to.

593 """

594 if (target := domain.ResolveWorkflow(match["stem"])) is None:

595 return ""

596

597 targetDocname, anchor = target

598 if (uri := sphinx.builder.get_relative_uri(docname, targetDocname)) == "": 598 ↛ 599line 598 didn't jump to line 599 because the condition on line 598 was never true

599 uri = PurePosixPath(sphinx.builder.get_target_uri(targetDocname)).name

600

601 return f'{match["space"]}URL="{uri}#{anchor}" target="_top"'

602

603 for node in doctree.findall(graphviz):

604 if CSS_CLASS in node["classes"]: 604 ↛ 603line 604 didn't jump to line 603 because the condition on line 604 was always true

605 node["code"] = LINK_MARKER.sub(link, node["code"])