git_utils.py

# Copyright (C) 2018 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the 'License');
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an 'AS IS' BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Helper functions to communicate with Git."""

import datetime
import re
import subprocess
from pathlib import Path

import fileutils
import hashtags
import reviewers
from manifest import Manifest

UNWANTED_TAGS = ["*alpha*", "*Alpha*", "*beta*", "*Beta*", "*rc*", "*RC*", "*test*"]


def fetch(proj_path: Path, remote_name: str, branch: str | None = None) -> None:
    """Runs git fetch.

    Args:
        proj_path: Path to Git repository.
        remote_name: A string to specify remote names.
    """
    cmd = ['git', 'fetch', '--tags', remote_name] + ([branch] if branch is not None else [])
    subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True)


def add_remote(proj_path: Path, name: str, url: str) -> None:
    """Adds a git remote.

    Args:
        proj_path: Path to Git repository.
        name: Name of the new remote.
        url: Url of the new remote.
    """
    cmd = ['git', 'remote', 'add', name, url]
    subprocess.run(cmd, cwd=proj_path, check=True)


def remove_remote(proj_path: Path, name: str) -> None:
    """Removes a git remote."""
    cmd = ['git', 'remote', 'remove', name]
    subprocess.run(cmd, cwd=proj_path, check=True)


def list_remotes(proj_path: Path) -> dict[str, str]:
    """Lists all Git remotes.

    Args:
        proj_path: Path to Git repository.

    Returns:
        A dict from remote name to remote url.
    """
    def parse_remote(line: str) -> tuple[str, str]:
        split = line.split()
        return split[0], split[1]

    cmd = ['git', 'remote', '-v']
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    lines = out.splitlines()
    return dict([parse_remote(line) for line in lines])


def detect_default_branch(proj_path: Path, remote_name: str) -> str:
    """Gets the name of the upstream's default branch to use."""
    cmd = ['git', 'remote', 'show', remote_name]
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    lines = out.splitlines()
    for line in lines:
        if "HEAD branch" in line:
            return line.split()[-1]
    raise RuntimeError(
        f"Could not find HEAD branch in 'git remote show {remote_name}'"
    )


def get_sha_for_branch(proj_path: Path, branch: str):
    """Gets the hash SHA for a branch."""
    cmd = ['git', 'rev-parse', branch]
    return subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                          text=True).stdout.strip()


def get_most_recent_tag(proj_path: Path, branch: str) -> str | None:
    """Finds the most recent tag that is reachable from HEAD."""
    cmd = ['git', 'describe', '--tags', branch, '--abbrev=0'] + \
          [f'--exclude={unwanted_tag}' for unwanted_tag in UNWANTED_TAGS]
    try:
        out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                            text=True).stdout.strip()
        return out
    except subprocess.CalledProcessError as ex:
        if "fatal: No names found" in ex.stderr:
            return None
        if "fatal: No tags can describe" in ex.stderr:
            return None
        raise


# pylint: disable=redefined-outer-name
def get_commit_time(proj_path: Path, commit: str) -> datetime.datetime:
    """Gets commit time of one commit."""
    cmd = ['git', 'show', '-s', '--format=%ct', commit]
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    return datetime.datetime.fromtimestamp(int(out.strip()))


def list_remote_branches(proj_path: Path, remote_name: str) -> list[str]:
    """Lists all branches for a remote."""
    cmd = ['git', 'branch', '-r']
    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                           text=True).stdout.splitlines()
    stripped = [line.strip() for line in lines]
    remote_path = remote_name + '/'
    return [
        line[len(remote_path):] for line in stripped
        if line.startswith(remote_path)
    ]


def list_local_branches(proj_path: Path) -> list[str]:
    """Lists all local branches."""
    cmd = ['git', 'branch', '--format=%(refname:short)']
    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                           text=True).stdout.splitlines()
    return lines


COMMIT_PATTERN = r'^[a-f0-9]{40}$'
COMMIT_RE = re.compile(COMMIT_PATTERN)


# pylint: disable=redefined-outer-name
def is_commit(commit: str) -> bool:
    """Whether a string looks like a SHA1 hash."""
    return bool(COMMIT_RE.match(commit))


def merge(proj_path: Path, branch: str) -> None:
    """Merges a branch."""
    try:
        cmd = ['git', 'merge', branch, '--no-commit']
        subprocess.run(cmd, cwd=proj_path, check=True)
    except subprocess.CalledProcessError as err:
        if hasattr(err, "output"):
            print(err.output)
        if not merge_conflict(proj_path):
            raise


def merge_conflict(proj_path: Path) -> bool:
    """Checks if there was a merge conflict."""
    cmd = ['git', 'ls-files', '--unmerged']
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    return bool(out)


def add_file(proj_path: Path, file_name: str) -> None:
    """Stages a file."""
    cmd = ['git', 'add', file_name]
    subprocess.run(cmd, cwd=proj_path, check=True)


def remove_gitmodules(proj_path: Path) -> None:
    """Deletes .gitmodules files."""
    cmd = ['find', '.', '-name', '.gitmodules', '-delete']
    subprocess.run(cmd, cwd=proj_path, check=True)


def delete_branch(proj_path: Path, branch_name: str) -> None:
    """Force delete a branch."""
    cmd = ['git', 'branch', '-D', branch_name]
    subprocess.run(cmd, cwd=proj_path, check=True)


def start_branch(proj_path: Path, branch_name: str) -> None:
    """Starts a new repo branch."""
    subprocess.run(['repo', 'start', branch_name], cwd=proj_path, check=True)


def repo_sync(proj_path: Path,) -> None:
    """Downloads new changes and updates the working files in the local environment."""
    subprocess.run(['repo', 'sync', '.'], cwd=proj_path, check=True)


def commit(proj_path: Path, message: str, no_verify: bool) -> None:
    """Commits changes."""
    cmd = ['git', 'commit', '-m', message] + (['--no-verify'] if no_verify is True else [])
    subprocess.run(cmd, cwd=proj_path, check=True)


def commit_amend(proj_path: Path) -> None:
    """Commits changes."""
    cmd = ['git', 'commit', '--amend', '--no-edit']
    subprocess.run(cmd, cwd=proj_path, check=True)


def checkout(proj_path: Path, branch_name: str) -> None:
    """Checkouts a branch."""
    cmd = ['git', 'checkout', branch_name]
    subprocess.run(cmd, cwd=proj_path, check=True)


def detach_to_android_head(proj_path: Path) -> None:
    """Detaches the project HEAD back to the manifest revision."""
    # -d detaches the project back to the manifest revision without updating.
    # -l avoids fetching new revisions from the remote. This might be superfluous with
    # -d, but I'm not sure, and it certainly doesn't harm anything.
    subprocess.run(['repo', 'sync', '-l', '-d', proj_path], cwd=proj_path, check=True)


def push(proj_path: Path, has_errors: bool) -> None:
    """Pushes change to remote."""
    remote_name = determine_remote_name(proj_path)
    cmd = ['git', 'push', remote_name, 'HEAD:refs/for/main', '-o', 'banned-words~skip']
    if revs := reviewers.find_reviewers(str(proj_path)):
        cmd.extend(['-o', revs])
    if tag := hashtags.find_hashtag(proj_path):
        cmd.extend(['-o', 't=' + tag])
    if has_errors:
        cmd.extend(['-o', 'l=Verified-1'])
    subprocess.run(cmd, cwd=proj_path, check=True)


def reset_hard(proj_path: Path) -> None:
    """Resets current HEAD and discards changes to tracked files."""
    cmd = ['git', 'reset', '--hard']
    subprocess.run(cmd, cwd=proj_path, check=True)


def clean(proj_path: Path) -> None:
    """Removes untracked files and directories."""
    cmd = ['git', 'clean', '-fdx']
    subprocess.run(cmd, cwd=proj_path, check=True)


def is_valid_url(proj_path: Path, url: str) -> bool:
    cmd = ['git', "ls-remote", url]
    return subprocess.run(cmd, cwd=proj_path, check=False, stdin=subprocess.DEVNULL,
                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                          start_new_session=True).returncode == 0


def list_remote_tags(proj_path: Path, remote_name: str) -> list[str]:
    """Lists tags in a remote repository."""
    cmd = ['git', "ls-remote", "--tags", remote_name]
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    lines = out.splitlines()
    return lines


def diff_stat(proj_path: Path, diff_filter: str, revision: str) -> str:
    try:
        cmd = ['git', 'diff', revision, '--stat', f'--diff-filter={diff_filter}']
        out = subprocess.run(cmd, capture_output=True, cwd=proj_path,
                             check=True, text=True).stdout
        return out
    except subprocess.CalledProcessError as err:
        return f"Could not calculate the diff: {err}"


def diff_name_only(proj_path: Path, diff_filter: str, revision: str) -> str:
    try:
        cmd = ['git', 'diff', revision, '--name-only', f'--diff-filter={diff_filter}']
        out = subprocess.run(cmd, capture_output=True, cwd=proj_path,
                             check=True, text=True).stdout
        return out
    except subprocess.CalledProcessError as err:
        return f"Could not calculate the diff: {err}"


def is_ancestor(proj_path: Path, ancestor: str, child: str) -> bool:
    cmd = ['git', 'merge-base', '--is-ancestor', ancestor, child]
    # https://git-scm.com/docs/git-merge-base#Documentation/git-merge-base.txt---is-ancestor
    # Exit status of 0 means yes, 1 means no, and all others mean an error occurred.
    # Although a commit is an ancestor of itself, we don't want to return True
    # if ancestor points to the same commit as child.
    if get_sha_for_branch(proj_path, ancestor) == child:
        return False
    try:
        subprocess.run(
            cmd,
            cwd=proj_path,
            text=True,
            stderr=subprocess.STDOUT,
            check=True,
            stdout=subprocess.PIPE
        )
        return True
    except subprocess.CalledProcessError as ex:
        if ex.returncode == 1:
            return False
        raise


def list_branches_with_commit(proj_path: Path, commit: str, remote_name: str) -> list[str]:
    """Lists upstream branches which contain the specified commit"""
    cmd = ['git', 'branch', '-r', '--contains', commit]
    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
                         text=True).stdout
    lines = out.splitlines()
    remote_branches = [line for line in lines if remote_name in line]
    return remote_branches


def determine_remote_name(proj_path: Path) -> str:
    """Returns the remote name in the manifest."""
    root = fileutils.find_tree_containing(proj_path)
    manifest = Manifest.for_tree(root)
    return manifest.remote