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

81GRAPH_ATTRIBUTES = (

82 "compound=true;",

83 "newrank=true;",

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

85 'fontsize="10";',

86 "nodesep=0.25;",

87 "ranksep=0.4;",

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

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

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

91)

92"""Attributes every pipeline graph is drawn with."""

94CSS_CLASS = "ghactions-pipeline-graph"

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

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

98"""

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

100replaces it by the link or removes it.

101"""

102

103_logger = getLogger(__name__)

104

105

106@export

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

108 """

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

110

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

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

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

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

115

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

117

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

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

120 repository and ref - it is never expanded;

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

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

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

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

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

126

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

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

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

130

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

132 in the direction the pipeline runs.

133

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

135 """

136

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

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

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

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

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

142

143 def __init__(

144 self,

145 workflow: Workflow,

146 resolver: Nullable[WorkflowResolver] = None,

147 direction: str = "LR",

148 depth: int = 0,

149 reduce: bool = True,

150 link: bool = True

151 ) -> None:

152 """

153 Initializes the graph by drawing a workflow.

154

155 :param workflow: The workflow to draw.

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

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

158 Default: a resolver knowing local references only.

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

160 bottom). Default: ``LR``.

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

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

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

164 :func:`resolveLinks`. Default: ``True``.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

180 """

181 from pyTooling.GitHub.WorkflowFile import Workflow, WorkflowResolver

182

183 if workflow is None:

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

185 elif not isinstance(workflow, Workflow):

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

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

188 raise ex

189

190 if resolver is None:

191 resolver = WorkflowResolver()

192 elif not isinstance(resolver, WorkflowResolver):

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

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

195 raise ex

196

197 if direction is None:

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

199 elif not isinstance(direction, str):

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

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

202 raise ex

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

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

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

206 raise ex

207

208 if depth is None:

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

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

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

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

213 raise ex

214 elif depth < 0:

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

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

217 raise ex

218

219 if reduce is None:

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

221 elif not isinstance(reduce, bool):

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

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

224 raise ex

225

226 if link is None:

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

228 elif not isinstance(link, bool):

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

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

231 raise ex

232

233 self._resolver = resolver

234 self._link = link

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

236 self._jobs = []

237 self._workflows = [workflow]

238

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

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

241

242 @readonly

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

244 """

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

246

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

248

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

250 """

251 return self._jobs

252

253 @readonly

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

255 """

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

257

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

259 """

260 return self._workflows

261

262 @staticmethod

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

264 """

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

266

267 :param text: The text to quote.

268 :returns: The text in double quotes.

269 """

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

271

272 return f'"{text}"'

273

274 @staticmethod

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

276 """

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

278

279 :param title: The title, unescaped.

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

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

282 """

283 label = html_escape(title, quote=False)

284 for line in lines:

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

286

287 return f"<{label}>"

288

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

290 """

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

292

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

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

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

296 """

297 lines = []

298 style = []

299 tooltip = []

300

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

302 lines.append(uses.FileName)

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

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

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

306

307 if isinstance(element, CIMatrix):

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

309 lines.append("matrix")

310 else:

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

312

313 if element.RuleCount > 0:

314 style.append("dashed")

315 for rule in element.Rules:

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

317

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

319

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

321 """

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

323

324 :param element: The element to draw.

325 :param identifier: Identifier of the node.

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

327 """

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

329

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

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

332 elif self._resolver.CanResolve(uses):

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

334 else:

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

336

337 if isinstance(element, CIMatrix):

338 attributes["peripheries"] = "2"

339

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

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

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

343

344 if tooltip != "":

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

346

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

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

349

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

351

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

353 """

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

355

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

357 :param cluster: Identifier of the cluster.

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

359 """

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

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

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

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

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

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

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

367 if tooltip != "":

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

369

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

371 """

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

373 matrix - and their edges.

374

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

376 round.

377

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

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

380 :param indent: Indentation of their statements.

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

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

383 """

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

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

386 for vertex in vertices:

387 element = vertex.Value

388 if element.Definition not in self._jobs:

389 self._jobs.append(element.Definition)

390

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

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

393 self._DrawNode(element, identifier, indent)

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

395 continue

396

397 cluster = f"cluster_{identifier}"

398 self._OpenCluster(element, cluster, indent)

399 if isinstance(element, CIWorkflow):

400 if element.CalledWorkflow not in self._workflows:

401 self._workflows.append(element.CalledWorkflow)

402

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

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

405

406 entryNode, exitNode = self._DrawVertices(

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

408 )

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

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

411

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

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

414 if prefix == "":

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

416 elif len(roots) > 1:

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

418

419 for vertex in vertices:

420 for edge in vertex.OutboundEdges:

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

422 head, _, headCluster = anchors[vertex]

423

424 attributes = []

425 if tailCluster is not None:

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

427

428 if headCluster is not None:

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

430

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

432 if len(attributes) > 0:

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

434

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

436

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

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

439

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

441

442 def __str__(self) -> str:

443 """

444 Render the graph.

445

446 :returns: The graph in the DOT language.

447 """

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

449

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

451

452

453@export

454class PipelineGraph(BaseDirective):

455 """

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

457

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

459

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

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

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

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

464

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

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

467 directory.

468 """

469

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

471

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

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

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

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

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

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

478 #: Mapping of option names to validator functions.

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

480 "align": align_spec,

481 "alt": strip,

482 "caption": strip,

483 "depth": directives.nonnegative_int,

484 "direction": strip,

485 "link": stripAndNormalize,

486 "name": strip,

487 "reduce": stripAndNormalize,

488 }

489

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

491 """

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

493

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

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

496

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

498 the workflow files are wrong.

499 """

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

501 self.env.note_dependency(relativePath)

502 workflowFile = Path(absolutePath)

503

504 from pyTooling.GitHub.WorkflowFile import WorkflowError, WorkflowResolver

505

506 repository = self.config.ghactions_repository

507 ref = self.config.ghactions_ref

508

509 try:

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

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

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

513 except SphinxExtensionError as ex:

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

515

516 try:

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

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

519 else:

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

521

522 graph = PipelineDotGraph(

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

524 )

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

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

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

528 message += f" {note}"

529

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

531

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

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

534

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

536 for job in graph.Jobs:

537 if (

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

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

540 ):

541 _logger.warning(

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

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

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

545 type="ghactions",

546 subtype="ref"

547 )

548

549 node = graphviz()

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

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

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

553 node["classes"] = [CSS_CLASS]

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

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

556

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

558 figure = figure_wrapper(self, node, caption)

559 self.add_name(figure)

560 return [figure]

561

562 self.add_name(node)

563 return [node]

564

565

566@export

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

568 """

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

570

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

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

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

574

575 :param sphinx: The Sphinx application.

576 :param doctree: The resolved document.

577 :param docname: Name of the document.

578 """

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

580 for node in doctree.findall(graphviz):

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

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

583

584 return

585

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

587

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

589 """

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

591

592 :param match: The marker.

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

594 when there is nothing to link to.

595 """

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

597 return ""

598

599 targetDocname, anchor = target

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

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

602

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

604

605 for node in doctree.findall(graphviz):

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

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