1# Copyright (C) 2018 The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the 'License');
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an 'AS IS' BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Helper functions to communicate with Git."""
15
16import datetime
17import re
18import subprocess
19from pathlib import Path
20from typing import Dict, List, Tuple
21
22import hashtags
23import reviewers
24
25def _run(cmd: List[str], cwd: Path) -> str:
26    """Runs a command and returns its output."""
27    return subprocess.check_output(cmd, text=True, cwd=cwd)
28
29
30def fetch(proj_path: Path, remote_names: List[str]) -> None:
31    """Runs git fetch.
32
33    Args:
34        proj_path: Path to Git repository.
35        remote_names: Array of string to specify remote names.
36    """
37    _run(['git', 'fetch', '--tags', '--multiple'] + remote_names, cwd=proj_path)
38
39
40def add_remote(proj_path: Path, name: str, url: str) -> None:
41    """Adds a git remote.
42
43    Args:
44        proj_path: Path to Git repository.
45        name: Name of the new remote.
46        url: Url of the new remote.
47    """
48    _run(['git', 'remote', 'add', name, url], cwd=proj_path)
49
50
51def remove_remote(proj_path: Path, name: str) -> None:
52    """Removes a git remote."""
53    _run(['git', 'remote', 'remove', name], cwd=proj_path)
54
55
56def list_remotes(proj_path: Path) -> Dict[str, str]:
57    """Lists all Git remotes.
58
59    Args:
60        proj_path: Path to Git repository.
61
62    Returns:
63        A dict from remote name to remote url.
64    """
65    def parse_remote(line: str) -> Tuple[str, str]:
66        split = line.split()
67        return (split[0], split[1])
68
69    out = _run(['git', 'remote', '-v'], proj_path)
70    lines = out.splitlines()
71    return dict([parse_remote(line) for line in lines])
72
73
74def get_sha_for_branch(proj_path: Path, branch: str):
75    """Gets the hash SHA for a branch."""
76    return _run(['git', 'rev-parse', branch], proj_path).strip()
77
78
79def get_commits_ahead(proj_path: Path, branch: str,
80                      base_branch: str) -> List[str]:
81    """Lists commits in `branch` but not `base_branch`."""
82    out = _run([
83        'git', 'rev-list', '--left-only', '--ancestry-path', '{}...{}'.format(
84            branch, base_branch)
85    ], proj_path)
86    return out.splitlines()
87
88
89# pylint: disable=redefined-outer-name
90def get_commit_time(proj_path: Path, commit: str) -> datetime.datetime:
91    """Gets commit time of one commit."""
92    out = _run(['git', 'show', '-s', '--format=%ct', commit], cwd=proj_path)
93    return datetime.datetime.fromtimestamp(int(out.strip()))
94
95
96def list_remote_branches(proj_path: Path, remote_name: str) -> List[str]:
97    """Lists all branches for a remote."""
98    lines = _run(['git', 'branch', '-r'], cwd=proj_path).splitlines()
99    stripped = [line.strip() for line in lines]
100    remote_path = remote_name + '/'
101    return [
102        line[len(remote_path):] for line in stripped
103        if line.startswith(remote_path)
104    ]
105
106
107def list_remote_tags(proj_path: Path, remote_name: str) -> List[str]:
108    """Lists all tags for a remote."""
109    regex = re.compile(r".*refs/tags/(?P<tag>[^\^]*).*")
110    def parse_remote_tag(line: str) -> str:
111        return regex.match(line).group("tag")
112
113    lines = _run(['git', "ls-remote", "--tags", remote_name],
114                 cwd=proj_path).splitlines()
115    tags = [parse_remote_tag(line) for line in lines]
116    return list(set(tags))
117
118
119def get_default_branch(proj_path: Path, remote_name: str) -> str:
120    """Gets the name of the upstream branch to use."""
121    branches_to_try = ['master', 'main']
122    remote_branches = list_remote_branches(proj_path, remote_name)
123    for branch in branches_to_try:
124        if branch in remote_branches:
125            return branch
126    # We couldn't find any of the branches we expected.
127    # Default to 'master', although nothing will work well.
128    return 'master'
129
130
131COMMIT_PATTERN = r'^[a-f0-9]{40}$'
132COMMIT_RE = re.compile(COMMIT_PATTERN)
133
134
135# pylint: disable=redefined-outer-name
136def is_commit(commit: str) -> bool:
137    """Whether a string looks like a SHA1 hash."""
138    return bool(COMMIT_RE.match(commit))
139
140
141def merge(proj_path: Path, branch: str) -> None:
142    """Merges a branch."""
143    try:
144        _run(['git', 'merge', branch, '--no-commit'], cwd=proj_path)
145    except subprocess.CalledProcessError:
146        # Merge failed. Error is already written to console.
147        _run(['git', 'merge', '--abort'], cwd=proj_path)
148        raise
149
150
151def add_file(proj_path: Path, file_name: str) -> None:
152    """Stages a file."""
153    _run(['git', 'add', file_name], cwd=proj_path)
154
155
156def delete_branch(proj_path: Path, branch_name: str) -> None:
157    """Force delete a branch."""
158    _run(['git', 'branch', '-D', branch_name], cwd=proj_path)
159
160
161def start_branch(proj_path: Path, branch_name: str) -> None:
162    """Starts a new repo branch."""
163    _run(['repo', 'start', branch_name], cwd=proj_path)
164
165
166def commit(proj_path: Path, message: str) -> None:
167    """Commits changes."""
168    _run(['git', 'commit', '-m', message], cwd=proj_path)
169
170
171def checkout(proj_path: Path, branch_name: str) -> None:
172    """Checkouts a branch."""
173    _run(['git', 'checkout', branch_name], cwd=proj_path)
174
175
176def push(proj_path: Path, remote_name: str, has_errors: bool) -> None:
177    """Pushes change to remote."""
178    cmd = ['git', 'push', remote_name, 'HEAD:refs/for/master']
179    if revs := reviewers.find_reviewers(str(proj_path)):
180        cmd.extend(['-o', revs])
181    if tag := hashtags.find_hashtag(proj_path):
182        cmd.extend(['-o', 't=' + tag])
183    if has_errors:
184        cmd.extend(['-o', 'l=Verified-1'])
185    _run(cmd, cwd=proj_path)
186