Skip to content
Back to top

# Source code for semantic_release.gitproject

    """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,
            force: bool = False,
            noop: bool = False,
        ) -> None:
            try:
                datetime.fromisoformat(isotimestamp)
            except ValueError as err:
                raise ValueError("Invalid timestamp format") from err

            if noop:
                command = str.join(
                    " ",
                    filter(
                        None,
                        [
                            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}'",
                            "--force" if force else "",
                        ],
                    ),
                ).strip()

                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(tag_name, a=True, m=message, force=force)
                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, force: bool = False
        ) -> None:
            if noop:
                noop_report(
                    indented(
                        f"""\
                        would have run:
                            git push {self._cred_masker.mask(remote_url)} tag {tag} {"--force" if force else ""}
                        """  # noqa: E501
                    )
                )
                return

            with Repo(str(self.project_root)) as repo:
                try:
                    repo.git.push(remote_url, "tag", tag, force=force)
                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(  # noqa: C901
            self,
            local_ref: str = "HEAD",
            upstream_ref: str = "origin",
            remote_url: str | None = None,
            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 upstream_ref: The name of the upstream remote or specific remote branch (default: origin)
            :param remote_url: Optional authenticated remote URL to use for fetching (default: None, uses configured remote)
            :param noop: Whether to skip the actual verification (for dry-run mode)

            :raises UpstreamBranchChangedError: If the upstream branch has changed
            """
            if not local_ref.strip():
                raise ValueError("Local reference cannot be empty")
            if not upstream_ref.strip():
                raise ValueError("Upstream reference cannot be empty")

            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 not None:
                    upstream_full_ref_name = tracking_branch.name
                    self.logger.info("Upstream branch name: %s", upstream_full_ref_name)
                else:
                    # If no tracking branch is set, derive it
                    upstream_name = (
                        upstream_ref.strip()
                        if upstream_ref.find("/") == -1
                        else upstream_ref.strip().split("/", maxsplit=1)[0]
                    )

                    if not repo.remotes or upstream_name not in repo.remotes:
                        err_msg = "No remote found; cannot verify upstream state!"
                        raise UnknownUpstreamBranchError(err_msg)

                    upstream_full_ref_name = (
                        f"{upstream_name}/{active_branch.name}"
                        if upstream_ref.find("/") == -1
                        else upstream_ref.strip()
                    )

                    if upstream_full_ref_name not in repo.refs:
                        err_msg = f"No upstream branch found for '{active_branch.name}'; cannot verify upstream state!"
                        raise UnknownUpstreamBranchError(err_msg)

                # 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:
                    # Check if we should use authenticated URL for fetch
                    # Only use remote_url if:
                    # 1. It's provided and different from the configured remote URL
                    # 2. It contains authentication credentials (@ symbol)
                    # 3. The configured remote is NOT a local path, file:// URL, or test URL (example.com)
                    #    This ensures we don't break tests or local development
                    configured_url = remote_ref_obj.url
                    is_local_or_test_remote = (
                        configured_url.startswith(("file://", "/", "C:/", "H:/"))
                        or "example.com" in configured_url
                        or not configured_url.startswith(
                            (
                                "https://",
                                "http://",
                                "git://",
                                "git@",
                                "ssh://",
                                "git+ssh://",
                            )
                        )
                    )

                    use_authenticated_fetch = (
                        remote_url
                        and "@" in remote_url
                        and remote_url != configured_url
                        and not is_local_or_test_remote
                    )

                    if use_authenticated_fetch:
                        # Use authenticated remote URL for fetch
                        # Fetch the remote branch and update the local tracking ref
                        repo.git.fetch(
                            remote_url,
                            f"refs/heads/{remote_branch_name}:refs/remotes/{upstream_full_ref_name}",
                        )
                    else:
                        # Use the default remote configuration for local paths,
                        # file:// URLs, test URLs, or when no authentication is needed
                        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,
                )