Coverage for pyTooling/Documentation/Sphinx/GitHubActions/Graph.py: 97%
254 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-03 23:02 +0000
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"""
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 .. gha: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 ``gha`` domain knows one.
46.. seealso::
48 :mod:`pyTooling.CI.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.Documentation.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.Documentation.Sphinx.Directives import BaseDirective, SphinxExtensionError, strip, stripAndNormalize
75if TYPE_CHECKING: # pragma: no cover
76 from pyTooling.CI.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 = "gha-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*)/\*gha-link:(?P<stem>[^*]*)\*/")
101_logger = getLogger(__name__)
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.
109 The workflow is converted by :meth:`Workflow.ToPipeline <pyTooling.CI.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.
114 The kind of an element is its node's shape and style:
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.
125 A job calling a reusable workflow that :class:`~pyTooling.CI.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.
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.
132 The graph is drawn in file order, so the same workflow is always drawn as the same DOT.
133 """
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.
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.
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.CI.GitHub.WorkflowFile.Workflow`.
165 :raises TypeError: If parameter 'resolver' is not of type
166 :class:`~pyTooling.CI.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.CI.GitHub.WorkflowFile import Workflow, WorkflowResolver
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
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
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
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
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
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
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]
237 graph = workflow.ToPipeline(resolver, depth).ToGraph(reduce=reduce)
238 self._DrawVertices(list(graph.IterateVertices()), "", "\t")
240 @readonly
241 def Jobs(self) -> list[Job]:
242 """
243 Read-only property to access the jobs drawn (:attr:`_jobs`).
245 A reusable workflow expanded twice is drawn twice, but its jobs are listed once.
247 :returns: The jobs of every workflow drawn, in the order they were drawn.
248 """
249 return self._jobs
251 @readonly
252 def Workflows(self) -> list[Workflow]:
253 """
254 Read-only property to access the workflows drawn (:attr:`_workflows`).
256 :returns: The entry-point workflow, followed by every reusable workflow expanded into a cluster, each once.
257 """
258 return self._workflows
260 @staticmethod
261 def _Quote(text: str) -> str:
262 """
263 Quote a text as a DOT string, keeping its line breaks as escape sequences.
265 :param text: The text to quote.
266 :returns: The text in double quotes.
267 """
268 text = text.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
270 return f'"{text}"'
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.
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>'
285 return f"<{label}>"
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.
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, one per line, or an empty string.
294 """
295 lines = []
296 style = []
297 tooltip = []
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}")
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)}")
311 if (condition := element.Condition) is not None:
312 style.append("dashed")
313 tooltip.append(f"if: {condition}")
315 return lines, style, "\n".join(tooltip)
317 def _DrawNode(self, element: CIBase, identifier: str, indent: str) -> None:
318 """
319 Draw a job, a matrix or a called workflow that isn't expanded as a node.
321 :param element: The element to draw.
322 :param identifier: Identifier of the node.
323 :param indent: Indentation of the node's statement.
324 """
325 lines, style, tooltip = self._Describe(element)
327 if (uses := element.Definition.Uses) is None:
328 attributes = {"style": ",".join(("filled", *style)), "fillcolor": "#f2f2f2"}
329 elif self._resolver.CanResolve(uses):
330 attributes = {"style": ",".join(("rounded", "filled", *style))}
331 else:
332 attributes = {"style": ",".join(("rounded", "filled", *style)), "fillcolor": "#ffffff"}
334 if isinstance(element, CIMatrix):
335 attributes["peripheries"] = "2"
337 statement = f"{indent}{self._Quote(identifier)} [label={self._Label(str(element), lines)}"
338 for name, value in attributes.items():
339 statement += f", {name}={self._Quote(value)}"
341 if tooltip != "":
342 statement += f", tooltip={self._Quote(tooltip)}"
344 if self._link and uses is not None and self._resolver.CanResolve(uses):
345 statement += f" /*gha-link:{uses.Stem}*/"
347 self._statements.append(f"{statement}];")
349 def _OpenCluster(self, element: CIBase, cluster: str, indent: str) -> None:
350 """
351 Open a cluster drawing a called workflow or a matrix: write its ``subgraph`` statement and its attributes.
353 :param element: The called workflow or the matrix.
354 :param cluster: Identifier of the cluster.
355 :param indent: Indentation of the cluster's statement.
356 """
357 lines, style, tooltip = self._Describe(element)
358 self._statements.append(f"{indent}subgraph {self._Quote(cluster)} {{")
359 self._statements.append(f"{indent}\tlabel={self._Label(str(element), lines)};")
360 self._statements.append(f'{indent}\tlabeljust="l";')
361 self._statements.append(f"{indent}\tstyle={self._Quote(','.join(('rounded', 'filled', *style)))};")
362 self._statements.append(f'{indent}\tcolor="#8a9bb2";')
363 self._statements.append(f'{indent}\tfillcolor="#f6f8fb";')
364 if tooltip != "":
365 self._statements.append(f"{indent}\ttooltip={self._Quote(tooltip)};")
367 def _DrawVertices(self, vertices: list[Vertex], prefix: str, indent: str) -> tuple[str, str]:
368 """
369 Draw the vertices of one level of the graph - the graph itself, or the subgraph of a called workflow or of a
370 matrix - and their edges.
372 An edge of the graph goes from the element needing to the element it needs; its arrow is drawn the other way
373 round.
375 :param vertices: The vertices, in file order.
376 :param prefix: Prefix of the identifiers of their nodes: the path of the jobs calling their workflow.
377 :param indent: Indentation of their statements.
378 :returns: Identifiers of the node an arrow into the level ends at - the first vertex needing nothing - and
379 of the node an arrow out of it starts at - the last vertex nothing needs.
380 """
381 # per vertex: the node an arrow into it ends at, the node an arrow out of it starts at, and its cluster
382 anchors: dict[Vertex, tuple[str, str, Nullable[str]]] = {}
383 for vertex in vertices:
384 element = vertex.Value
385 if element.Definition not in self._jobs:
386 self._jobs.append(element.Definition)
388 identifier = f"{prefix}{element}"
389 if not isinstance(element, (CIWorkflow, CIMatrix)) or vertex.OutboundLinkCount == 0:
390 self._DrawNode(element, identifier, indent)
391 anchors[vertex] = (identifier, identifier, None)
392 continue
394 cluster = f"cluster_{identifier}"
395 self._OpenCluster(element, cluster, indent)
396 if isinstance(element, CIWorkflow):
397 if element.CalledWorkflow not in self._workflows:
398 self._workflows.append(element.CalledWorkflow)
400 if self._link: 400 ↛ 403line 400 didn't jump to line 403 because the condition on line 400 was always true
401 self._statements.append(f"{indent}\t/*gha-link:{element.Definition.Uses.Stem}*/")
403 entryNode, exitNode = self._DrawVertices(
404 [link.Destination for link in vertex.OutboundLinks], f"{identifier}/", f"{indent}\t"
405 )
406 self._statements.append(f"{indent}}}")
407 anchors[vertex] = (entryNode, exitNode, cluster)
409 # the vertices needing nothing start together: in the first rank of the graph, or side by side in their cluster
410 roots = [self._Quote(anchors[vertex][0]) for vertex in vertices if vertex.OutboundEdgeCount == 0]
411 if prefix == "":
412 self._statements.append(f"{indent}{{rank=min; {'; '.join(roots)};}}")
413 elif len(roots) > 1:
414 self._statements.append(f"{indent}{{rank=same; {'; '.join(roots)};}}")
416 for vertex in vertices:
417 for edge in vertex.OutboundEdges:
418 _, tail, tailCluster = anchors[edge.Destination]
419 head, _, headCluster = anchors[vertex]
421 attributes = []
422 if tailCluster is not None:
423 attributes.append(f"ltail={self._Quote(tailCluster)}")
425 if headCluster is not None:
426 attributes.append(f"lhead={self._Quote(headCluster)}")
428 statement = f"{indent}{self._Quote(tail)} -> {self._Quote(head)}"
429 if len(attributes) > 0:
430 statement += f" [{', '.join(attributes)}]"
432 self._statements.append(f"{statement};")
434 first = next(vertex for vertex in vertices if vertex.OutboundEdgeCount == 0)
435 last = [vertex for vertex in vertices if vertex.InboundEdgeCount == 0][-1]
437 return anchors[first][0], anchors[last][1]
439 def __str__(self) -> str:
440 """
441 Render the graph.
443 :returns: The graph in the DOT language.
444 """
445 statements = "\n".join(self._statements)
447 return f"digraph pipeline {{\n{statements}\n}}"
450@export
451class PipelineGraph(BaseDirective):
452 """
453 The ``gha:pipeline-graph`` directive: the pipeline of a workflow, drawn from the workflow file.
455 One argument, the path of the workflow file relative to the document using the directive.
457 The documented repository and the directory of its workflow files are the configuration values ``gha_repository``
458 and ``gha_workflow_directory`` - relative to the source directory, and by default the directory of the drawn
459 workflow file. A job calling a reusable workflow of that repository is expanded, while ``:depth:`` allows, and
460 checked against the ref ``gha_ref``, if that is configured.
462 The workflow files are read by the resolver of the ``gha`` domain, which reads every file once per build. Without
463 ``gha_workflow_directory``, a resolver of its own maps ``gha_repository`` to the drawn file's directory.
464 """
466 directiveName: str = "gha:pipeline-graph" #: Name the directive is invoked by.
468 has_content = False #: A boolean; ``True`` if content is allowed.
469 required_arguments = 1 #: Number of required directive arguments: the workflow file's path.
470 optional_arguments = 0 #: Number of optional arguments after the required ones.
471 final_argument_whitespace = False #: A boolean; ``True`` if the last argument may contain spaces.
472 # docutils declares 'option_spec' on 'Directive' and 'BaseDirective' assigns it, so mypy calls every
473 # spelling of this override a conflict with one of them
474 #: Mapping of option names to validator functions.
475 option_spec: dict[str, Any] = { # type: ignore[misc]
476 "align": align_spec,
477 "alt": strip,
478 "caption": strip,
479 "depth": directives.nonnegative_int,
480 "direction": strip,
481 "link": stripAndNormalize,
482 "name": strip,
483 "reduce": stripAndNormalize,
484 }
486 def run(self) -> list[nodes.Node]:
487 """
488 Read the workflow file and hand its pipeline to :mod:`sphinx.ext.graphviz` for rendering.
490 Every workflow file drawn becomes a dependency of the document. A job calling a reusable workflow of the
491 documented repository at another ref than ``gha_ref`` is reported as a warning of type ``gha.ref``.
493 :returns: A ``graphviz`` node, wrapped in a figure when a caption was given, or an error node when the options or
494 the workflow files are wrong.
495 """
496 relativePath, absolutePath = self.env.relfn2path(self.arguments[0])
497 self.env.note_dependency(relativePath)
498 workflowFile = Path(absolutePath)
500 from pyTooling.CI.GitHub.WorkflowFile import WorkflowError, WorkflowResolver
502 repository = self.config.gha_repository
503 ref = self.config.gha_ref
505 try:
506 direction = self._ParseStringOption("direction", "LR", "(?i)(LR|TB)$").upper()
507 reduce = self._ParseBooleanOption("reduce", True)
508 link = self._ParseBooleanOption("link", True)
509 except SphinxExtensionError as ex:
510 return [self.state.document.reporter.error(str(ex), line=self.lineno)]
512 try:
513 if repository is not None and self.config.gha_workflow_directory is None:
514 resolver = WorkflowResolver({repository: workflowFile.parent})
515 else:
516 resolver = self.env.get_domain("gha").Resolver
518 graph = PipelineDotGraph(
519 resolver.Load(workflowFile), resolver, direction, self.options.get("depth", 0), reduce, link
520 )
521 except (OSError, ValueError, WorkflowError) as ex:
522 message = f"{self.directiveName}: Couldn't draw '{self.arguments[0]}': {ex}"
523 for note in getattr(ex, "__notes__", []):
524 message += f" {note}"
526 return [self.state.document.reporter.error(message, line=self.lineno)]
528 for workflow in graph.Workflows[1:]:
529 self.env.note_dependency(str(workflow.Path))
531 if repository is not None and ref is not None:
532 for job in graph.Jobs:
533 if (
534 (uses := job.Uses) is not None and uses.Repository is not None and
535 uses.Repository.lower() == repository.lower() and uses.Reference != ref
536 ):
537 _logger.warning(
538 f"{self.directiveName}: Job '{job.Name}' calls '{uses.FileName}' at ref '{uses.Reference}', but the "
539 f"documentation describes ref '{ref}' ({uses.Location}).",
540 location=(self.env.docname, self.lineno),
541 type="gha",
542 subtype="ref"
543 )
545 node = graphviz()
546 node["code"] = str(graph)
547 node["options"] = {"docname": self.env.docname}
548 node["alt"] = self.options.get("alt", f"Pipeline of {workflowFile.name}")
549 node["classes"] = [CSS_CLASS]
550 if "align" in self.options: 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 node["align"] = self.options["align"]
553 if (caption := self.options.get("caption", None)) is not None: 553 ↛ 554line 553 didn't jump to line 554 because the condition on line 553 was never true
554 figure = figure_wrapper(self, node, caption)
555 self.add_name(figure)
556 return [figure]
558 self.add_name(node)
559 return [node]
562@export
563def resolveLinks(sphinx: Sphinx, doctree: nodes.document, docname: str) -> None:
564 """
565 Call-back for Sphinx' ``doctree-resolved`` event, linking the jobs of a pipeline graph to their reusable workflows.
567 A job calling a reusable workflow of the documented repository links to the page documenting that workflow, as the
568 ``gha`` domain resolves the workflow's file stem with ``ResolveWorkflow``. Without a page for the workflow, or in a
569 format other than HTML, the job has no link.
571 :param sphinx: The Sphinx application.
572 :param doctree: The resolved document.
573 :param docname: Name of the document.
574 """
575 if sphinx.builder.format != "html":
576 for node in doctree.findall(graphviz):
577 if CSS_CLASS in node["classes"]: 577 ↛ 576line 577 didn't jump to line 576 because the condition on line 577 was always true
578 node["code"] = LINK_MARKER.sub("", node["code"])
580 return
582 domain = sphinx.env.get_domain("gha")
584 def link(match: Match[str]) -> str:
585 """
586 Nested function returning the link attributes replacing a marker, or nothing.
588 :param match: The marker.
589 :returns: The whitespace before the marker and the attributes ``URL`` and ``target``, or an empty string
590 when there is nothing to link to.
591 """
592 if (target := domain.ResolveWorkflow(match["stem"])) is None:
593 return ""
595 targetDocname, anchor = target
596 if (uri := sphinx.builder.get_relative_uri(docname, targetDocname)) == "": 596 ↛ 597line 596 didn't jump to line 597 because the condition on line 596 was never true
597 uri = PurePosixPath(sphinx.builder.get_target_uri(targetDocname)).name
599 return f'{match["space"]}URL="{uri}#{anchor}" target="_top"'
601 for node in doctree.findall(graphviz):
602 if CSS_CLASS in node["classes"]: 602 ↛ 601line 602 didn't jump to line 601 because the condition on line 602 was always true
603 node["code"] = LINK_MARKER.sub(link, node["code"])