gitproject.py

"""Module for git related operations."""

from __future__ import annotations

from contextlib import nullcontext
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING

from git import GitCommandError, Repo

from semantic_release.cli.masking_filter import MaskingFilter
from semantic_release.cli.util import indented, noop_report
from semantic_release.errors import (
    DetachedHeadGitError,
    GitAddError,
    GitCommitEmptyIndexError,
    GitCommitError,
    GitFetchError,
    GitPushError,
    GitTagError,
    LocalGitError,
    UnknownUpstreamBranchError,
    UpstreamBranchChangedError,
)
from semantic_release.globals import logger

if TYPE_CHECKING:  # pragma: no cover
    from contextlib import _GeneratorContextManager
    from logging import Logger
    from typing import Sequence

    from git import Actor


class GitProject:
    def __init__(
        self,
        directory: Path | str = ".",
        commit_author: Actor | None = None,
        credential_masker: MaskingFilter | None = None,
    ) -> None:
        self._project_root = Path(directory).resolve()
        self._logger = logger
        self._cred_masker = credential_masker or MaskingFilter()
        self._commit_author = commit_author

    @property
    def project_root(self) -> Path:
        return self._project_root

    @property
    def logger(self) -> Logger:
        return self._logger

    def _get_custom_environment(
        self,
        repo: Repo,
        custom_vars: dict[str, str] | None = None,
    ) -> nullcontext[None] | _GeneratorContextManager[None]:
        """
        git.custom_environment is a context manager but
        is not reentrant, so once we have "used" it
        we need to throw it away and re-create it in
        order to use it again
        """
        author_vars = (
            {
                "GIT_AUTHOR_NAME": self._commit_author.name,
                "GIT_AUTHOR_EMAIL": self._commit_author.email,
                "GIT_COMMITTER_NAME": self._commit_author.name,
                "GIT_COMMITTER_EMAIL": self._commit_author.email,
            }
            if self._commit_author
            else {}
        )

        custom_env_vars = {
            **author_vars,
            **(custom_vars or {}),
        }

        return (
            nullcontext()
            if not custom_env_vars
            else repo.git.custom_environment(**custom_env_vars)
        )

    def is_dirty(self) -> bool:
        with Repo(str(self.project_root)) as repo:
            return repo.is_dirty()

    def is_shallow_clone(self) -> bool:
        """
        Check if the repository is a shallow clone.

        :return: True if the repository is a shallow clone, False otherwise
        """
        with Repo(str(self.project_root)) as repo:
            shallow_file = Path(repo.git_dir, "shallow")
            return shallow_file.exists()

    def git_unshallow(self, noop: bool = False) -> None:
        """
        Convert a shallow clone to a full clone by fetching the full history.

        :param noop: Whether or not to actually run the unshallow command
        """
        if noop:
            noop_report("would have run:\n" "    git fetch --unshallow")
            return

        with Repo(str(self.project_root)) as repo:
            try:
                self.logger.info("Converting shallow clone to full clone...")
                repo.git.fetch("--unshallow")
                self.logger.info("Repository unshallowed successfully")
            except GitCommandError as err:
                # If the repository is already a full clone, git fetch --unshallow will fail
                # with "fatal: --unshallow on a complete repository does not make sense"
                # We can safely ignore this error by checking the stderr message
                stderr = str(err.stderr) if err.stderr else ""
                if "does not make sense" in stderr or "complete repository" in stderr:
                    self.logger.debug("Repository is already a full clone")
                else:
                    self.logger.exception(str(err))
                    raise

    def git_add(
        self,
        paths: Sequence[Path | str],
        force: bool = False,
        strict: bool = False,
        noop: bool = False,
    ) -> None:
        if noop:
            noop_report(
                indented(
                    f"""\
                    would have run:
                        git add {str.join(" ", [str(Path(p)) for p in paths])}
                    """
                )
            )
            return

        git_args = dict(
            filter(
                lambda k_v: k_v[1],  # if truthy
                {
                    "force": force,
                }.items(),
            )
        )

        with Repo(str(self.project_root)) as repo:
            # TODO: in future this loop should be 1 line:
            # repo.index.add(all_paths_to_add, force=False)  # noqa: ERA001
            # but since 'force' is deliberately ineffective (as in docstring) in gitpython 3.1.18
            # we have to do manually add each filepath, and catch the exception if it is an ignored file
            for updated_path in paths:
                try:
                    repo.git.add(str(Path(updated_path)), **git_args)
                except GitCommandError as err:  # noqa: PERF203, acceptable performance loss
                    err_msg = f"Failed to add path ({updated_path}) to index"
                    if strict:
                        self.logger.exception(str(err))
                        raise GitAddError(err_msg) from err
                    self.logger.warning(err_msg)

    def git_commit(
        self,
        message: str,
        date: int | None = None,
        commit_all: bool = False,
        no_verify: bool = False,
        noop: bool = False,
    ) -> None:
        git_args = dict(
            filter(
                lambda k_v: k_v[1],  # if truthy
                {
                    "a": commit_all,
                    "m": message,
                    "date": date,
                    "no_verify": no_verify,
                }.items(),
            )
        )

        if noop:
            command = (
                f"""\
                GIT_AUTHOR_NAME={self._commit_author.name} \\
                GIT_AUTHOR_EMAIL={self._commit_author.email} \\
                GIT_COMMITTER_NAME={self._commit_author.name} \\
                GIT_COMMITTER_EMAIL={self._commit_author.email} \\
                """
                if self._commit_author
                else ""
            )

            # Indents the newlines so that terminal formatting is happy - note the
            # git commit line of the output is 24 spaces indented too
            # Only this message needs such special handling because of the newlines
            # that might be in a commit message between the subject and body
            indented_commit_message = message.replace("\n\n", "\n\n" + " " * 24)

            command += f"git commit -m '{indented_commit_message}'"
            command += "--all" if commit_all else ""
            command += "--no-verify" if no_verify else ""

            noop_report(
                indented(
                    f"""\
                    would have run:
                        {command}
                    """
                )
            )
            return

        with Repo(str(self.project_root)) as repo:
            has_index_changes = bool(repo.index.diff("HEAD"))
            has_working_changes = self.is_dirty()
            will_commit_files = has_index_changes or (
                has_working_changes and commit_all
            )

            if not will_commit_files:
                raise GitCommitEmptyIndexError("No changes to commit!")

            with self._get_custom_environment(repo):
                try:
                    repo.git.commit(**git_args)
                except GitCommandError as err:
                    self.logger.exception(str(err))
                    raise GitCommitError("Failed to commit changes") from err

    def git_tag(
        self, tag_name: str, message: str, isotimestamp: str, noop: bool = False
    ) -> None:
        try:
            datetime.fromisoformat(isotimestamp)
        except ValueError as err:
            raise ValueError("Invalid timestamp format") from err

        if noop:
            command = str.join(
                " ",
                [
                    f"GIT_COMMITTER_DATE={isotimestamp}",
                    *(
                        [
                            f"GIT_AUTHOR_NAME={self._commit_author.name}",
                            f"GIT_AUTHOR_EMAIL={self._commit_author.email}",
                            f"GIT_COMMITTER_NAME={self._commit_author.name}",
                            f"GIT_COMMITTER_EMAIL={self._commit_author.email}",
                        ]
                        if self._commit_author
                        else [""]
                    ),
                    f"git tag -a {tag_name} -m '{message}'",
                ],
            )

            noop_report(
                indented(
                    f"""\
                    would have run:
                        {command}
                    """
                )
            )
            return

        with Repo(str(self.project_root)) as repo, self._get_custom_environment(
            repo,
            {"GIT_COMMITTER_DATE": isotimestamp},
        ):
            try:
                repo.git.tag("-a", tag_name, m=message)
            except GitCommandError as err:
                self.logger.exception(str(err))
                raise GitTagError(f"Failed to create tag ({tag_name})") from err

    def git_push_branch(self, remote_url: str, branch: str, noop: bool = False) -> None:
        if noop:
            noop_report(
                indented(
                    f"""\
                    would have run:
                        git push {self._cred_masker.mask(remote_url)} {branch}
                    """
                )
            )
            return

        with Repo(str(self.project_root)) as repo:
            try:
                repo.git.push(remote_url, branch)
            except GitCommandError as err:
                self.logger.exception(str(err))
                raise GitPushError(
                    f"Failed to push branch ({branch}) to remote"
                ) from err

    def git_push_tag(self, remote_url: str, tag: str, noop: bool = False) -> None:
        if noop:
            noop_report(
                indented(
                    f"""\
                    would have run:
                        git push {self._cred_masker.mask(remote_url)} tag {tag}
                    """  # noqa: E501
                )
            )
            return

        with Repo(str(self.project_root)) as repo:
            try:
                repo.git.push(remote_url, "tag", tag)
            except GitCommandError as err:
                self.logger.exception(str(err))
                raise GitPushError(f"Failed to push tag ({tag}) to remote") from err

    def verify_upstream_unchanged(
        self, local_ref: str = "HEAD", noop: bool = False
    ) -> None:
        """
        Verify that the upstream branch has not changed since the given local reference.

        :param local_ref: The local reference to compare against upstream (default: HEAD)
        :param noop: Whether to skip the actual verification (for dry-run mode)

        :raises UpstreamBranchChangedError: If the upstream branch has changed
        """
        if noop:
            noop_report(
                indented(
                    """\
                    would have verified that upstream branch has not changed
                    """
                )
            )
            return

        with Repo(str(self.project_root)) as repo:
            # Get the current active branch
            try:
                active_branch = repo.active_branch
            except TypeError:
                # When in detached HEAD state, active_branch raises TypeError
                err_msg = (
                    "Repository is in detached HEAD state, cannot verify upstream state"
                )
                raise DetachedHeadGitError(err_msg) from None

            # Get the tracking branch (upstream branch)
            if (tracking_branch := active_branch.tracking_branch()) is None:
                err_msg = f"No upstream branch found for '{active_branch.name}'; cannot verify upstream state!"
                raise UnknownUpstreamBranchError(err_msg)

            upstream_full_ref_name = tracking_branch.name
            self.logger.info("Upstream branch name: %s", upstream_full_ref_name)

            # Extract the remote name from the tracking branch
            # tracking_branch.name is in the format "remote/branch"
            remote_name, remote_branch_name = upstream_full_ref_name.split(
                "/", maxsplit=1
            )
            remote_ref_obj = repo.remotes[remote_name]

            # Fetch the latest changes from the remote
            self.logger.info("Fetching latest changes from remote '%s'", remote_name)
            try:
                remote_ref_obj.fetch()
            except GitCommandError as err:
                self.logger.exception(str(err))
                err_msg = f"Failed to fetch from remote '{remote_name}'"
                raise GitFetchError(err_msg) from err

            # Get the SHA of the upstream branch
            try:
                upstream_commit_ref = remote_ref_obj.refs[remote_branch_name].commit
                upstream_sha = upstream_commit_ref.hexsha
            except AttributeError as err:
                self.logger.exception(str(err))
                err_msg = f"Unable to determine upstream branch SHA for '{upstream_full_ref_name}'"
                raise GitFetchError(err_msg) from err

            # Get the SHA of the specified ref (default: HEAD)
            try:
                local_commit = repo.commit(repo.git.rev_parse(local_ref))
            except GitCommandError as err:
                self.logger.exception(str(err))
                err_msg = f"Unable to determine the SHA for local ref '{local_ref}'"
                raise LocalGitError(err_msg) from err

            # Compare the two SHAs
            if local_commit.hexsha != upstream_sha and not any(
                commit.hexsha == upstream_sha for commit in local_commit.iter_parents()
            ):
                err_msg = str.join(
                    "\n",
                    (
                        f"[LOCAL SHA] {local_commit.hexsha} != {upstream_sha} [UPSTREAM SHA].",
                        f"Upstream branch '{upstream_full_ref_name}' has changed!",
                    ),
                )
                raise UpstreamBranchChangedError(err_msg)

            self.logger.info(
                "Verified upstream branch '%s' has not changed",
                upstream_full_ref_name,
            )