Implement updating of Github Actions.

scons/github_actions.py

@@ -6,6 +6,7 @@
 import sys
 
 from dataclasses import dataclass, field
+from functools import reduce
 from glob import glob
 from os import environ, path
 from typing import List, Optional, Set
@@ -15,11 +16,7 @@
 
 ENV_GITHUB_TOKEN = 'GITHUB_TOKEN'
 
-SEPARATOR_VERSION = '@'
-
-SEPARATOR_VERSION_NUMBER = '.'
-
-SEPARATOR_PATH = '/'
+WORKFLOW_ENCODING = 'utf-8'
 
 
 @dataclass
@@ -32,20 +29,39 @@ class ActionVersion:
     """
     version: str
 
+    SEPARATOR = '.'
+
+    @staticmethod
+    def from_version_numbers(*version_numbers: int) -> 'ActionVersion':
+        """
+        Creates and returns the version of a GitHub Action from one or several version numbers.
+
+        :param version_numbers: The version numbers
+        :return:                The version that has been created
+        """
+        return ActionVersion(ActionVersion.SEPARATOR.join([str(version_number) for version_number in version_numbers]))
+
+    @property
+    def version_numbers(self) -> List[int]:
+        """
+        A list that stores the individual version numbers, the full version consists of.
+        """
+        return [int(version_number) for version_number in str(self).split(self.SEPARATOR)]
+
     def __str__(self) -> str:
         return self.version.lstrip('v')
 
     def __lt__(self, other: 'ActionVersion') -> bool:
-        first_numbers = str(self).split(SEPARATOR_VERSION_NUMBER)
-        second_numbers = str(other).split(SEPARATOR_VERSION_NUMBER)
+        first_version_numbers = self.version_numbers
+        second_version_numbers = other.version_numbers
 
-        for i in range(min(len(first_numbers), len(second_numbers))):
-            first = int(first_numbers[i])
-            second = int(second_numbers[i])
+        for i in range(min(len(first_version_numbers), len(second_version_numbers))):
+            first_version_number = first_version_numbers[i]
+            second_version_number = second_version_numbers[i]
 
-            if first > second:
+            if first_version_number > second_version_number:
                 return False
-            if first < second:
+            if first_version_number < second_version_number:
                 return True
 
         return False
@@ -65,18 +81,21 @@ class Action:
     version: ActionVersion
     latest_version: Optional[ActionVersion] = None
 
+    SEPARATOR = '@'
+
     @staticmethod
-    def parse(uses: str) -> 'Action':
+    def from_uses_clause(uses_clause: str) -> 'Action':
         """
-        Parses and returns a GitHub Action as specified via the uses-clause of a workflow.
+        Creates and returns a GitHub Action from the uses-clause of a workflow.
 
-        :param uses:    The uses-clause
-        :return:        The GitHub Action
+        :param uses_clause: The uses-clause
+        :return:            The GitHub Action that has been created
         """
-        parts = uses.split(SEPARATOR_VERSION)
+        parts = uses_clause.split(Action.SEPARATOR)
 
         if len(parts) != 2:
-            raise ValueError('Action must contain the symbol + "' + SEPARATOR_VERSION + '", but got "' + uses + '"')
+            raise ValueError('Uses-clause must contain the symbol + "' + Action.SEPARATOR + '", but got "' + uses_clause
+                             + '"')
 
         return Action(name=parts[0], version=ActionVersion(parts[1]))
 
@@ -86,19 +105,19 @@ def repository(self) -> str:
         The name of the repository, where the GitHub Action is hosted.
         """
         repository = self.name
-        parts = repository.split(SEPARATOR_PATH)
-        return SEPARATOR_PATH.join(parts[:2]) if len(parts) > 2 else repository
+        separator = '/'
+        parts = repository.split(separator)
+        return separator.join(parts[:2]) if len(parts) > 2 else repository
 
+    @property
     def is_outdated(self) -> bool:
         """
-        Returns whether the GitHub Action is known to be outdated or not.
-
-        :return: True, if the GitHub Action is outdated, False otherwise
+        True, if the GitHub Action is known to be outdated, False otherwise.
         """
         return self.latest_version and self.version < self.latest_version
 
     def __str__(self) -> str:
-        return self.name + SEPARATOR_VERSION + str(self.version)
+        return self.name + self.SEPARATOR + str(self.version)
 
     def __eq__(self, other: 'Action') -> bool:
         return str(self) == str(other)
@@ -114,46 +133,95 @@ class Workflow:
 
     Attributes:
         workflow_file:  The path of the workflow definition file
-        actions:        A set that stores the Actions in the workflow
+        yaml_dict:      A dictionary that stores the YAML structure of the workflow definition file
+        actions:        A set that stores all Actions in the workflow
     """
     workflow_file: str
+    yaml_dict: dict
     actions: Set[Action] = field(default_factory=set)
 
+    TAG_USES = 'uses'
+
+    @property
+    def uses_clauses(self) -> List[str]:
+        """
+        A list that contains all uses-clauses in the workflow.
+        """
+        uses_clauses = []
+
+        for job in self.yaml_dict.get('jobs', {}).values():
+            for step in job.get('steps', []):
+                uses_clause = step.get(self.TAG_USES, None)
+
+                if uses_clause:
+                    uses_clauses.append(uses_clause)
+
+        return uses_clauses
+
+    @property
+    def outdated_actions(self) -> Set[Action]:
+        """
+        A set that stores all Actions in the workflow that are known to be outdated.
+        """
+        return {action for action in self.actions if action.is_outdated}
+
     def __eq__(self, other: 'Workflow') -> bool:
         return self.workflow_file == other.workflow_file
 
     def __hash__(self):
         return hash(self.workflow_file)
 
 
-def __get_github_workflow_files(directory: str) -> List[str]:
-    return glob(path.join(directory, '*.y*ml'))
-
-
-def __load_yaml(workflow_file: str) -> dict:
+def __read_workflow(workflow_file: str) -> Workflow:
     install_build_dependencies('pyyaml')
     # pylint: disable=import-outside-toplevel
     import yaml
-    with open(workflow_file, encoding='utf-8') as file:
-        return yaml.load(file.read(), Loader=yaml.CLoader)
+    with open(workflow_file, mode='r', encoding=WORKFLOW_ENCODING) as file:
+        yaml_dict = yaml.load(file.read(), Loader=yaml.CLoader)
+        return Workflow(workflow_file=workflow_file, yaml_dict=yaml_dict)
+
+
+def __read_workflow_lines(workflow_file: str) -> List[str]:
+    with open(workflow_file, mode='r', encoding=WORKFLOW_ENCODING) as file:
+        return file.readlines()
+
+
+def __write_workflow_lines(workflow_file: str, lines: List[str]):
+    with open(workflow_file, mode='w', encoding=WORKFLOW_ENCODING) as file:
+        file.writelines(lines)
+
+
+def __update_workflow(workflow_file: str, *updated_actions: Action):
+    updated_actions_by_name = reduce(lambda aggr, x: dict(aggr, **{x.name: x}), updated_actions, {})
+    lines = __read_workflow_lines(workflow_file)
+    uses_prefix = Workflow.TAG_USES + ':'
+    updated_lines = []
+
+    for line in lines:
+        updated_lines.append(line)
+        line_stripped = line.strip()
+
+        if line_stripped.startswith(uses_prefix):
+            uses_clause = line_stripped[len(uses_prefix):].strip()
+            action = Action.from_uses_clause(uses_clause)
+            updated_action = updated_actions_by_name.get(action.name)
+
+            if updated_action:
+                updated_lines[-1] = line.replace(str(action.version), str(updated_action.version))
+
+    __write_workflow_lines(workflow_file, updated_lines)
 
 
 def __parse_workflow(workflow_file: str) -> Workflow:
     print('Searching for GitHub Actions in workflow "' + workflow_file + '"...')
-    workflow = Workflow(workflow_file)
-    workflow_yaml = __load_yaml(workflow_file)
+    workflow = __read_workflow(workflow_file)
 
-    for job in workflow_yaml.get('jobs', {}).values():
-        for step in job.get('steps', []):
-            uses = step.get('uses', None)
-
-            if uses:
-                try:
-                    action = Action.parse(uses)
-                    workflow.actions.add(action)
-                except ValueError as error:
-                    print('Failed to parse uses-clause in workflow "' + workflow_file + '": ' + str(error))
-                    sys.exit(-1)
+    for uses_clause in workflow.uses_clauses:
+        try:
+            workflow.actions.add(Action.from_uses_clause(uses_clause))
+        except ValueError as error:
+            print('Failed to parse uses-clause in workflow "' + workflow_file + '": ' + str(error))
+            sys.exit(-1)
 
     return workflow
 
@@ -208,36 +276,79 @@ def __determine_latest_action_versions(*workflows: Workflow) -> Set[Workflow]:
     return set(workflows)
 
 
+def __parse_all_workflows() -> Set[Workflow]:
+    workflow_directory = path.join('.github', 'workflows')
+    workflow_files = glob(path.join(workflow_directory, '*.y*ml'))
+    return __determine_latest_action_versions(*__parse_workflows(*workflow_files))
+
+
+def __print_table(header: List[str], rows: List[List[str]]):
+    install_build_dependencies('tabulate')
+    # pylint: disable=import-outside-toplevel
+    from tabulate import tabulate
+    print(tabulate(rows, headers=header))
+
+
 def __print_outdated_actions(*workflows: Workflow):
     rows = []
 
     for workflow in workflows:
-        for action in workflow.actions:
-            if action.is_outdated():
-                rows.append([workflow.workflow_file, str(action.name), str(action.version), str(action.latest_version)])
+        for action in workflow.outdated_actions:
+            rows.append([workflow.workflow_file, str(action.name), str(action.version), str(action.latest_version)])
 
     if rows:
         rows.sort(key=lambda row: (row[0], row[1]))
         header = ['Workflow', 'Action', 'Current version', 'Latest version']
-        install_build_dependencies('tabulate')
-        # pylint: disable=import-outside-toplevel
-        from tabulate import tabulate
         print('The following GitHub Actions are outdated:\n')
-        print(tabulate(rows, headers=header))
+        __print_table(header=header, rows=rows)
+    else:
+        print('All GitHub Actions are up-to-date!')
+
+
+def __update_outdated_actions(*workflows: Workflow) -> Set[Workflow]:
+    rows = []
+
+    for workflow in workflows:
+        outdated_actions = workflow.outdated_actions
+
+        if outdated_actions:
+            workflow_file = workflow.workflow_file
+            updated_actions = set()
+
+            for action in outdated_actions:
+                previous_version = action.version
+                previous_version_numbers = previous_version.version_numbers
+                latest_version_numbers = action.latest_version.version_numbers
+                max_version_numbers = min(len(previous_version_numbers), len(latest_version_numbers))
+                updated_version = ActionVersion.from_version_numbers(*latest_version_numbers[:max_version_numbers])
+                rows.append([workflow_file, action.name, str(previous_version), str(updated_version)])
+                action.version = updated_version
+                updated_actions.add(action)
+
+            __update_workflow(workflow_file, *updated_actions)
+
+    if rows:
+        rows.sort(key=lambda row: (row[0], row[1]))
+        header = ['Workflow', 'Action', 'Previous version', 'Updated version']
+        print('The following GitHub Actions have been updated:\n')
+        __print_table(header=header, rows=rows)
+    else:
+        print('No GitHub Actions have been updated.')
+
+    return set(workflows)
 
 
 def check_github_actions(**_):
     """
     Checks the project's GitHub workflows for outdated Actions.
     """
-    workflow_directory = path.join('.github', 'workflows')
-    workflow_files = __get_github_workflow_files(workflow_directory)
-    workflows = __determine_latest_action_versions(*__parse_workflows(*workflow_files))
+    workflows = __parse_all_workflows()
     __print_outdated_actions(*workflows)
 
 
 def update_github_actions(**_):
     """
     Updates the versions of outdated GitHub Actions in the project's workflows.
     """
-    print('Updating versions of outdated GitHub Actions...')
+    workflows = __parse_all_workflows()
+    __update_outdated_actions(*workflows)