Feature/698 update a workflow by yaml

.import_linter_config

@@ -28,3 +28,4 @@ ignore_imports =
     # in each CLI usage), we allow the noxconfig to be imported within these modules.
     exasol.toolbox.nox.* -> noxconfig
     exasol.toolbox.tools.template -> noxconfig
+    exasol.toolbox.util.workflow.* -> noxconfig

exasol/toolbox/util/workflows/patch_workflow.py

@@ -1,17 +1,33 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
 from enum import Enum
+from functools import cached_property
 from pathlib import Path
-from typing import Any
+from typing import (
+    Annotated,
+    Any,
+    TypeAlias,
+)
 
 from pydantic import (
     BaseModel,
     ConfigDict,
     Field,
+    ValidationError,
 )
 from ruamel.yaml import CommentedMap
 
 from exasol.toolbox.util.workflows.render_yaml import YamlRenderer
 
 
+class InvalidWorkflowPatcherYamlError(Exception):
+    def __init__(self, file_path: Path):
+        super().__init__(
+            f"File '{file_path}' is malformed; it failed Pydantic validation."
+        )
+
+
 class ActionType(str, Enum):
     INSERT_AFTER = "INSERT_AFTER"
     REPLACE = "REPLACE"
@@ -82,15 +98,51 @@ class WorkflowPatcherConfig(BaseModel):
     workflows: list[Workflow]
 
 
+WorkflowCommentedMap: TypeAlias = Annotated[
+    CommentedMap, f"This CommentedMap is structured according to `{Workflow.__name__}`"
+]
+
+StepCustomizationCommentedMap: TypeAlias = Annotated[
+    CommentedMap,
+    f"This CommentedMap is structured according to `{StepCustomization.__name__}`",
+]
+StepsCommentedMap: TypeAlias = Annotated[
+    CommentedMap,
+    f"This CommentedMap is structured according to `list[{StepContent.__name__}]`",
+]
+
+
+@dataclass(frozen=True)
 class WorkflowPatcher(YamlRenderer):
     """
     The :class:`WorkflowPatcher` enables users to define a YAML file
-    to customize PTB-provided workflows by removing or modifying jobs in the file.
-    A job can be modified by replacing or inserting steps.
-    The provided YAML file must meet the conditions of :class:`WorkflowPatcherConfig`.
+    to customize PTB-provided workflows by removing or modifying jobs.
+    A job can be modified by replacing or inserting steps. The provided
+    YAML file must meet the conditions of :class:`WorkflowPatcherConfig`.
     """
 
-    def get_yaml_dict(self, file_path: Path) -> CommentedMap:
-        loaded_yaml = super().get_yaml_dict(file_path)
-        WorkflowPatcherConfig.model_validate(loaded_yaml)
-        return loaded_yaml
+    @cached_property
+    def content(self) -> CommentedMap:
+        """
+        The loaded YAML content. It loads on first access and
+        stays cached even though the class is frozen.
+        """
+        loaded_yaml = self.get_yaml_dict()
+        try:
+            WorkflowPatcherConfig.model_validate(loaded_yaml)
+            return loaded_yaml
+        except ValidationError as exc:
+            raise InvalidWorkflowPatcherYamlError(file_path=self.file_path) from exc
+
+    def extract_by_workflow(self, workflow_name: str) -> WorkflowCommentedMap | None:
+        """
+        Extract from the `content` where `name` matches the `workflow_name`.
+        If the workflow is not found, then `None` is returned. It is an expected and
+        common use case that the `WorkflowPatcher` would only modify a few workflows
+        and not all of them.
+        """
+        inner_content = self.content["workflows"]
+        for workflow in inner_content:
+            if workflow["name"] == workflow_name:
+                return workflow
+        return None

exasol/toolbox/util/workflows/process_template.py

@@ -1,8 +1,18 @@
-from pathlib import Path
+import copy
+from dataclasses import dataclass
 
+from ruamel.yaml import CommentedMap
+
+from exasol.toolbox.util.workflows.patch_workflow import (
+    ActionType,
+    StepCustomizationCommentedMap,
+    StepsCommentedMap,
+    WorkflowCommentedMap,
+)
 from exasol.toolbox.util.workflows.render_yaml import YamlRenderer
 
 
+@dataclass(frozen=True)
 class WorkflowRenderer(YamlRenderer):
     """
     The :class:`WorkflowRenderer` renders a workflow template provided by the PTB into
@@ -11,9 +21,90 @@ class WorkflowRenderer(YamlRenderer):
       - standardizing formatting via ruamel.yaml for a consistent output.
     """
 
-    def render(self, file_path: Path) -> str:
+    patch_yaml: WorkflowCommentedMap | None = None
+
+    def render(self) -> str:
         """
         Render the template to the contents of a valid GitHub workflow.
         """
-        workflow_dict = self.get_yaml_dict(file_path)
+        workflow_dict = self.get_yaml_dict()
+
+        if self.patch_yaml:
+            workflow_modifier = WorkflowPatcher(
+                workflow_dict=workflow_dict, patch_yaml=self.patch_yaml
+            )
+            workflow_dict = workflow_modifier.get_patched_workflow()
+
         return self.get_as_string(workflow_dict)
+
+
+@dataclass(frozen=True)
+class WorkflowPatcher:
+    workflow_dict: WorkflowCommentedMap
+    patch_yaml: CommentedMap
+
+    def __post_init__(self):
+        # Perform deepcopy to ensure this instance owns its data
+        # Use object.__setattr__ because the dataclass is frozen
+        object.__setattr__(self, "workflow_dict", copy.deepcopy(self.workflow_dict))
+
+    @property
+    def jobs_dict(self) -> CommentedMap:
+        return self.workflow_dict["jobs"]
+
+    def _get_step_list(self, job_name: str) -> StepsCommentedMap:
+        self._verify_job_exists(job_name=job_name)
+        return self.jobs_dict[job_name]["steps"]
+
+    def _customize_steps(
+        self, step_customizations: StepCustomizationCommentedMap
+    ) -> None:
+        """
+        Customize the steps of jobs specified in `step_customizations` in a workflow
+        (`workflow_dict`). If a `step_id` or its parent `job` cannot be found, an
+        exception is raised.
+        """
+        for patch in step_customizations:
+            job_name = patch["job"]
+            idx = self._get_step_index(job_name=job_name, step_id=patch["step_id"])
+
+            step_list = self._get_step_list(job_name=job_name)
+            if patch["action"] == ActionType.REPLACE.value:
+                step_list[idx : idx + 1] = patch["content"]
+            elif patch["action"] == ActionType.INSERT_AFTER.value:
+                step_list[idx + 1 : idx + 1] = patch["content"]
+
+    def _get_step_index(self, job_name: str, step_id: str) -> int:
+        steps = self._get_step_list(job_name=job_name)
+        for index, step in enumerate(steps):
+            if step["id"] == step_id:
+                return index
+        raise ValueError(f"step_id '{step_id}' not found in job '{job_name}'")
+
+    def _remove_jobs(self, remove_jobs: CommentedMap) -> None:
+        """
+        Remove the jobs specified in `remove_jobs` in a workflow yaml (`workflow_dict`).
+        If a `job` cannot be found, an exception is raised.
+        """
+        for job_name in remove_jobs:
+            self._verify_job_exists(job_name)
+            self.jobs_dict.pop(job_name)
+
+    def _verify_job_exists(self, job_name: str) -> None:
+        if job_name not in self.jobs_dict:
+            raise ValueError(f"job '{job_name}' not found")
+
+    def get_patched_workflow(self):
+        """
+        Patch the `workflow_dict`. As dictionaries are mutable structures, we directly
+        take advantage of this by having it modified in this class's internal methods
+        without explicit returns.
+        """
+
+        if remove_jobs := self.patch_yaml.get("remove_jobs", {}):
+            self._remove_jobs(remove_jobs=remove_jobs)
+
+        if step_customizations := self.patch_yaml.get("step_customizations", {}):
+            self._customize_steps(step_customizations=step_customizations)
+
+        return self.workflow_dict

exasol/toolbox/util/workflows/render_yaml.py

@@ -4,17 +4,46 @@
 from pathlib import Path
 from typing import Any
 
-from jinja2 import Environment
+from jinja2 import (
+    Environment,
+    StrictUndefined,
+    TemplateError,
+)
 from ruamel.yaml import (
     YAML,
     CommentedMap,
 )
+from ruamel.yaml.error import YAMLError
 
 jinja_env = Environment(
-    variable_start_string="((", variable_end_string="))", autoescape=True
+    variable_start_string="((",
+    variable_end_string="))",
+    autoescape=True,
+    # This requires that all Jinja variables must be defined in the provided
+    # dictionary. If not, then a `jinja2.exceptions.UndefinedError` exception
+    # will be raised.
+    undefined=StrictUndefined,
 )
 
 
+class YamlSyntaxError(Exception):
+    """Raised when the rendered template is not a valid YAML document."""
+
+    def __init__(self, file_path: Path):
+        super().__init__(
+            f"File '{file_path}' could not be parsed by ruamel-yaml. Check for invalid YAML syntax."
+        )
+
+
+class TemplateRenderingError(Exception):
+    """Raised when Jinja2 fails to process the template."""
+
+    def __init__(self, file_path: Path):
+        super().__init__(
+            f"File '{file_path}' failed to render. Check Jinja2-related errors."
+        )
+
+
 @dataclass(frozen=True)
 class YamlRenderer:
     """
@@ -25,6 +54,7 @@ class YamlRenderer:
     """
 
     github_template_dict: dict[str, Any]
+    file_path: Path
 
     @staticmethod
     def _get_standard_yaml() -> YAML:
@@ -45,18 +75,22 @@ def _render_with_jinja(self, input_str: str) -> str:
         jinja_template = jinja_env.from_string(input_str)
         return jinja_template.render(self.github_template_dict)
 
-    def get_yaml_dict(self, file_path: Path) -> CommentedMap:
+    def get_yaml_dict(self) -> CommentedMap:
         """
         Load a file as a CommentedMap (dictionary form of a YAML), after
         rendering it with Jinja.
         """
-        with file_path.open("r", encoding="utf-8") as stream:
+        with self.file_path.open("r", encoding="utf-8") as stream:
             raw_content = stream.read()
 
-        workflow_string = self._render_with_jinja(raw_content)
-
-        yaml = self._get_standard_yaml()
-        return yaml.load(workflow_string)
+        try:
+            workflow_string = self._render_with_jinja(raw_content)
+            yaml = self._get_standard_yaml()
+            return yaml.load(workflow_string)
+        except TemplateError as exc:
+            raise TemplateRenderingError(file_path=self.file_path) from exc
+        except YAMLError as exc:
+            raise YamlSyntaxError(file_path=self.file_path) from exc
 
     def get_as_string(self, yaml_dict: CommentedMap) -> str:
         """

exasol/toolbox/util/workflows/workflow.py

@@ -19,11 +19,7 @@ def load_from_template(cls, file_path: Path, github_template_dict: dict[str, Any
         if not file_path.exists():
             raise FileNotFoundError(file_path)
 
-        try:
-            workflow_renderer = WorkflowRenderer(
-                github_template_dict=github_template_dict
-            )
-            workflow = workflow_renderer.render(file_path=file_path)
-            return cls(content=workflow)
-        except Exception as e:
-            raise ValueError(f"Error rendering file: {file_path}") from e
+        workflow_renderer = WorkflowRenderer(
+            github_template_dict=github_template_dict, file_path=file_path
+        )
+        return cls(content=workflow_renderer.render())

test/unit/util/workflows/conftest.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass
+from inspect import cleandoc
+from pathlib import Path
+
+import pytest
+
+from exasol.toolbox.util.workflows.patch_workflow import WorkflowPatcher
+from noxconfig import PROJECT_CONFIG
+
+
+@dataclass(frozen=True)
+class ExamplePatcherYaml:
+    remove_jobs = """
+        workflows:
+        - name: "checks.yml"
+          remove_jobs:
+            - build-documentation-and-check-links
+        """
+    step_customization = """
+        workflows:
+        - name: "checks.yml"
+          step_customizations:
+            - action: {action}
+              job: run-unit-tests
+              step_id: check-out-repository
+              content:
+                - name: Check out Repository
+                  id: check-out-repository
+                  uses: actions/checkout@v6
+                  with:
+                    fetch-depth: 0
+        """
+
+
+@pytest.fixture(scope="session")
+def example_patcher_yaml():
+    return ExamplePatcherYaml
+
+
+@pytest.fixture
+def workflow_patcher_yaml(tmp_path: Path) -> Path:
+    return tmp_path / ".workflow-patcher.yml"
+
+
+@pytest.fixture
+def workflow_patcher(workflow_patcher_yaml) -> WorkflowPatcher:
+    return WorkflowPatcher(
+        github_template_dict=PROJECT_CONFIG.github_template_dict,
+        file_path=workflow_patcher_yaml,
+    )
+
+
+@pytest.fixture
+def remove_job_yaml(example_patcher_yaml, workflow_patcher_yaml):
+    content = cleandoc(example_patcher_yaml.remove_jobs)
+    workflow_patcher_yaml.write_text(content)
+    return content
+
+
+@pytest.fixture
+def step_customization_yaml(request, example_patcher_yaml, workflow_patcher_yaml):
+    # request.param will hold the value passed from @pytest.mark.parametrize
+    action_value = request.param
+
+    text = example_patcher_yaml.step_customization.format(action=action_value)
+    content = cleandoc(text)
+    workflow_patcher_yaml.write_text(content)
+    return content