Sign commits

src/commit.py

@@ -0,0 +1,108 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Module for handling interactions with git commit."""
+
+import re
+from collections.abc import Iterator
+from itertools import takewhile
+from pathlib import Path
+from typing import NamedTuple
+
+
+class FileAdded(NamedTuple):
+    """File that was added or copied in a commit.
+
+    Attributes:
+        path: The location of the file on disk.
+        content: The content of the file.
+    """
+
+    path: Path
+    content: str
+
+
+class FileModified(NamedTuple):
+    """File that was modified in a commit.
+
+    Attributes:
+        path: The location of the file on disk.
+        content: The content of the file.
+    """
+
+    path: Path
+    content: str
+
+
+class FileDeleted(NamedTuple):
+    """File that was deleted in a commit.
+
+    Attributes:
+        path: The location of the file on disk.
+    """
+
+    path: Path
+
+
+# Copied will be mapped to added and renamed will be mapped to be a delete and add
+FileAction = FileAdded | FileModified | FileDeleted
+_ADDED_PATTERN = re.compile(r"A\s*(\S*)")
+_MODIFIED_PATTERN = re.compile(r"M\s*(\S*)")
+_DELETED_PATTERN = re.compile(r"D\s*(\S*)")
+_RENAMED_PATTERN = re.compile(r"R\d+\s*(\S*)\s*(\S*)")
+_COPIED_PATTERN = re.compile(r"C\d+\s*(\S*)\s*(\S*)")
+
+
+def parse_git_show(output: str, repository_path: Path) -> Iterator[FileAction]:
+    """Parse the output of a git show with --name-status into manageable data.
+
+    Args:
+        output: The output of the git show command.
+        repository_path: The path to the git repository.
+
+    Yields:
+        Information about each of the files that changed in the commit.
+    """
+    # Processing in reverse up to empty line to detect end of file changes as an empty line.
+    # Example git show output:
+    #     git show --name-status <commit sha>
+    #     commit <commit sha> (HEAD -> <branch name>)
+    #     Author: <author>
+    #     Date:   <date>
+
+    #         <commit message>
+
+    #     A       add-file.text
+    #     M       change-file.text
+    #     D       delete-file.txt
+    #     R100    renamed-file.text       is-renamed-file.text
+    #     C100    to-be-copied-file.text  copied-file.text
+    # The copied example is a guess, was not able to get the copied status during testing
+    lines = takewhile(bool, reversed(output.splitlines()))
+    for line in lines:
+        if (added_match := _ADDED_PATTERN.match(line)) is not None:
+            path = Path(added_match.group(1))
+            yield FileAdded(path, (repository_path / path).read_text(encoding="utf-8"))
+            continue
+
+        if (copied_match := _COPIED_PATTERN.match(line)) is not None:
+            path = Path(copied_match.group(2))
+            yield FileAdded(path, (repository_path / path).read_text(encoding="utf-8"))
+            continue
+
+        if (modified_match := _MODIFIED_PATTERN.match(line)) is not None:
+            path = Path(modified_match.group(1))
+            yield FileModified(path, (repository_path / path).read_text(encoding="utf-8"))
+            continue
+
+        if (delete_match := _DELETED_PATTERN.match(line)) is not None:
+            path = Path(delete_match.group(1))
+            yield FileDeleted(path)
+            continue
+
+        if (renamed_match := _RENAMED_PATTERN.match(line)) is not None:
+            old_path = Path(renamed_match.group(1))
+            path = Path(renamed_match.group(2))
+            yield FileDeleted(old_path)
+            yield FileAdded(path, (repository_path / path).read_text(encoding="utf-8"))
+            continue

src/repository.py

@@ -6,25 +6,27 @@
 import base64
 import logging
 import re
-from collections.abc import Iterator, Sequence
+from collections.abc import Iterable, Iterator, Sequence
 from contextlib import contextmanager
 from functools import cached_property
 from itertools import chain
 from pathlib import Path
-from typing import Any, NamedTuple
+from typing import Any, NamedTuple, cast
 
 from git import GitCommandError
 from git.diff import Diff
 from git.repo import Repo
 from github import Github
 from github.GithubException import GithubException, UnknownObjectException
+from github.InputGitTreeElement import InputGitTreeElement
 from github.PullRequest import PullRequest
 from github.Repository import Repository
 
 from src.docs_directory import has_docs_directory
 from src.metadata import get as get_metadata
 from src.types_ import Metadata
 
+from . import commit as commit_module
 from .constants import DOCUMENTATION_FOLDER_NAME
 from .exceptions import (
     InputError,
@@ -133,6 +135,39 @@ def __str__(self) -> str:
         return " // ".join(chain(modified_str, new_str, removed_str))
 
 
+def _commit_file_to_tree_element(commit_file: commit_module.FileAction) -> InputGitTreeElement:
+    """Convert a file with an action to a tree element.
+
+    Args:
+        commit_file: The file action to convert.
+
+    Returns:
+        The git tree element.
+
+    Raises:
+        NotImplementedError: for unsupported commit file types.
+    """
+    match type(commit_file):
+        case commit_module.FileAdded:
+            commit_file = cast(commit_module.FileAdded, commit_file)
+            return InputGitTreeElement(
+                path=str(commit_file.path), mode="100644", type="blob", content=commit_file.content
+            )
+        case commit_module.FileModified:
+            commit_file = cast(commit_module.FileModified, commit_file)
+            return InputGitTreeElement(
+                path=str(commit_file.path), mode="100644", type="blob", content=commit_file.content
+            )
+        case commit_module.FileDeleted:
+            commit_file = cast(commit_module.FileDeleted, commit_file)
+            return InputGitTreeElement(
+                path=str(commit_file.path), mode="100644", type="blob", sha=None
+            )
+        # Here just in case, should not occur in production
+        case _:  # pragma: no cover
+            raise NotImplementedError(f"unsupported file in commit, {commit_file}")
+
+
 class Client:
     """Wrapper for git/git-server related functionalities.
 
@@ -320,6 +355,25 @@ def create_branch(self, branch_name: str, base: str | None = None) -> "Client":
 
         return self
 
+    def _github_client_push(
+        self, commit_files: Iterable[commit_module.FileAction], commit_msg: str
+    ) -> None:
+        """Push files from a commit to GitHub using PyGithub.
+
+        Args:
+            commit_files: The files that were added, modified or deleted in a commit.
+            commit_msg: The message to use for commits.
+        """
+        branch = self._github_repo.get_branch(self.current_branch)
+        current_tree = self._github_repo.get_git_tree(sha=branch.commit.sha)
+        tree_elements = [_commit_file_to_tree_element(commit_file) for commit_file in commit_files]
+        tree = self._github_repo.create_git_tree(tree_elements, current_tree)
+        commit = self._github_repo.create_git_commit(
+            message=commit_msg, tree=tree, parents=[branch.commit.commit]
+        )
+        branch_git_ref = self._github_repo.get_git_ref(f"heads/{self.current_branch}")
+        branch_git_ref.edit(sha=commit.sha)
+
     def update_branch(
         self,
         commit_msg: str,
@@ -342,15 +396,37 @@ def update_branch(
         Returns:
             Repository client with the updated branch
         """
+        push_args = ["-u"]
+        if force:
+            push_args.append("-f")
+        push_args.extend([ORIGIN_NAME, self.current_branch])
+
         try:
+            # Create the branch if it doesn't exist
+            if push:
+                self._git_repo.git.push(*push_args)
+
             self._git_repo.git.add("-A", directory or ".")
             self._git_repo.git.commit("-m", f"'{commit_msg}'")
             if push:
-                args = ["-u"]
-                if force:
-                    args.append("-f")
-                args.extend([ORIGIN_NAME, self.current_branch])
-                self._git_repo.git.push(*args)
+                try:
+                    self._git_repo.git.push(*push_args)
+                except GitCommandError as exc:
+                    # Try with the PyGithub client, suppress any errors and report the original
+                    # problem on failure
+                    try:
+                        logging.info(
+                            "encountered error with push, try to use GitHub API to sign commits"
+                        )
+                        show_output = self._git_repo.git.show("--name-status")
+                        commit_files = commit_module.parse_git_show(
+                            output=show_output, repository_path=self.base_path
+                        )
+                        self._github_client_push(commit_files=commit_files, commit_msg=commit_msg)
+                    except (GitCommandError, GithubException) as nested_exc:
+                        # Raise original exception, flake8-docstrings-complete confuses this with a
+                        # specific exception rather than re-raising
+                        raise exc from nested_exc  # noqa: DCO053
         except GitCommandError as exc:
             raise RepositoryClientError(
                 f"Unexpected error updating branch {self.current_branch}. {exc=!r}"