• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
20
21import hashtags
22import reviewers
23
24UNWANTED_TAGS = ["*alpha*", "*Alpha*", "*beta*", "*Beta*", "*rc*", "*RC*", "*test*"]
25
26
27def fetch(proj_path: Path, remote_name: str, branch: str | None = None) -> None:
28    """Runs git fetch.
29
30    Args:
31        proj_path: Path to Git repository.
32        remote_name: A string to specify remote names.
33    """
34    cmd = ['git', 'fetch', '--tags', remote_name] + ([branch] if branch is not None else [])
35    subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True)
36
37
38def add_remote(proj_path: Path, name: str, url: str) -> None:
39    """Adds a git remote.
40
41    Args:
42        proj_path: Path to Git repository.
43        name: Name of the new remote.
44        url: Url of the new remote.
45    """
46    cmd = ['git', 'remote', 'add', name, url]
47    subprocess.run(cmd, cwd=proj_path, check=True)
48
49
50def remove_remote(proj_path: Path, name: str) -> None:
51    """Removes a git remote."""
52    cmd = ['git', 'remote', 'remove', name]
53    subprocess.run(cmd, cwd=proj_path, check=True)
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    cmd = ['git', 'remote', '-v']
70    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
71                         text=True).stdout
72    lines = out.splitlines()
73    return dict([parse_remote(line) for line in lines])
74
75
76def detect_default_branch(proj_path: Path, remote_name: str) -> str:
77    """Gets the name of the upstream's default branch to use."""
78    cmd = ['git', 'remote', 'show', remote_name]
79    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
80                         text=True).stdout
81    lines = out.splitlines()
82    for line in lines:
83        if "HEAD branch" in line:
84            return line.split()[-1]
85    raise RuntimeError(
86        f"Could not find HEAD branch in 'git remote show {remote_name}'"
87    )
88
89
90def get_sha_for_branch(proj_path: Path, branch: str):
91    """Gets the hash SHA for a branch."""
92    cmd = ['git', 'rev-parse', branch]
93    return subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
94                          text=True).stdout.strip()
95
96
97def get_most_recent_tag(proj_path: Path, branch: str) -> str | None:
98    """Finds the most recent tag that is reachable from HEAD."""
99    cmd = ['git', 'describe', '--tags', branch, '--abbrev=0'] + \
100          [f'--exclude={unwanted_tag}' for unwanted_tag in UNWANTED_TAGS]
101    try:
102        out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
103                            text=True).stdout.strip()
104        return out
105    except subprocess.CalledProcessError as ex:
106        if "fatal: No names found" in ex.stderr:
107            return None
108        if "fatal: No tags can describe" in ex.stderr:
109            return None
110        raise
111
112
113# pylint: disable=redefined-outer-name
114def get_commit_time(proj_path: Path, commit: str) -> datetime.datetime:
115    """Gets commit time of one commit."""
116    cmd = ['git', 'show', '-s', '--format=%ct', commit]
117    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
118                         text=True).stdout
119    return datetime.datetime.fromtimestamp(int(out.strip()))
120
121
122def list_remote_branches(proj_path: Path, remote_name: str) -> list[str]:
123    """Lists all branches for a remote."""
124    cmd = ['git', 'branch', '-r']
125    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
126                           text=True).stdout.splitlines()
127    stripped = [line.strip() for line in lines]
128    remote_path = remote_name + '/'
129    return [
130        line[len(remote_path):] for line in stripped
131        if line.startswith(remote_path)
132    ]
133
134
135def list_local_branches(proj_path: Path) -> list[str]:
136    """Lists all local branches."""
137    cmd = ['git', 'branch', '--format=%(refname:short)']
138    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
139                           text=True).stdout.splitlines()
140    return lines
141
142
143COMMIT_PATTERN = r'^[a-f0-9]{40}$'
144COMMIT_RE = re.compile(COMMIT_PATTERN)
145
146
147# pylint: disable=redefined-outer-name
148def is_commit(commit: str) -> bool:
149    """Whether a string looks like a SHA1 hash."""
150    return bool(COMMIT_RE.match(commit))
151
152
153def merge(proj_path: Path, branch: str) -> None:
154    """Merges a branch."""
155    try:
156        cmd = ['git', 'merge', branch, '--no-commit']
157        subprocess.run(cmd, cwd=proj_path, check=True)
158    except subprocess.CalledProcessError as err:
159        if hasattr(err, "output"):
160            print(err.output)
161        if not merge_conflict(proj_path):
162            raise
163
164
165def merge_conflict(proj_path: Path) -> bool:
166    """Checks if there was a merge conflict."""
167    cmd = ['git', 'ls-files', '--unmerged']
168    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
169                         text=True).stdout
170    return bool(out)
171
172
173def add_file(proj_path: Path, file_name: str) -> None:
174    """Stages a file."""
175    cmd = ['git', 'add', file_name]
176    subprocess.run(cmd, cwd=proj_path, check=True)
177
178
179def remove_gitmodules(proj_path: Path) -> None:
180    """Deletes .gitmodules files."""
181    cmd = ['find', '.', '-name', '.gitmodules', '-delete']
182    subprocess.run(cmd, cwd=proj_path, check=True)
183
184
185def delete_branch(proj_path: Path, branch_name: str) -> None:
186    """Force delete a branch."""
187    cmd = ['git', 'branch', '-D', branch_name]
188    subprocess.run(cmd, cwd=proj_path, check=True)
189
190
191def start_branch(proj_path: Path, branch_name: str) -> None:
192    """Starts a new repo branch."""
193    subprocess.run(['repo', 'start', branch_name], cwd=proj_path, check=True)
194
195
196def repo_sync(proj_path: Path,) -> None:
197    """Downloads new changes and updates the working files in the local environment."""
198    subprocess.run(['repo', 'sync', '.'], cwd=proj_path, check=True)
199
200
201def commit(proj_path: Path, message: str, no_verify: bool) -> None:
202    """Commits changes."""
203    cmd = ['git', 'commit', '-m', message] + (['--no-verify'] if no_verify is True else [])
204    subprocess.run(cmd, cwd=proj_path, check=True)
205
206
207def commit_amend(proj_path: Path) -> None:
208    """Commits changes."""
209    cmd = ['git', 'commit', '--amend', '--no-edit']
210    subprocess.run(cmd, cwd=proj_path, check=True)
211
212
213def checkout(proj_path: Path, branch_name: str) -> None:
214    """Checkouts a branch."""
215    cmd = ['git', 'checkout', branch_name]
216    subprocess.run(cmd, cwd=proj_path, check=True)
217
218
219def detach_to_android_head(proj_path: Path) -> None:
220    """Detaches the project HEAD back to the manifest revision."""
221    # -d detaches the project back to the manifest revision without updating.
222    # -l avoids fetching new revisions from the remote. This might be superfluous with
223    # -d, but I'm not sure, and it certainly doesn't harm anything.
224    subprocess.run(['repo', 'sync', '-l', '-d', proj_path], cwd=proj_path, check=True)
225
226
227def push(proj_path: Path, remote_name: str, has_errors: bool) -> None:
228    """Pushes change to remote."""
229    cmd = ['git', 'push', remote_name, 'HEAD:refs/for/main', '-o', 'banned-words~skip']
230    if revs := reviewers.find_reviewers(str(proj_path)):
231        cmd.extend(['-o', revs])
232    if tag := hashtags.find_hashtag(proj_path):
233        cmd.extend(['-o', 't=' + tag])
234    if has_errors:
235        cmd.extend(['-o', 'l=Verified-1'])
236    subprocess.run(cmd, cwd=proj_path, check=True)
237
238
239def reset_hard(proj_path: Path) -> None:
240    """Resets current HEAD and discards changes to tracked files."""
241    cmd = ['git', 'reset', '--hard']
242    subprocess.run(cmd, cwd=proj_path, check=True)
243
244
245def clean(proj_path: Path) -> None:
246    """Removes untracked files and directories."""
247    cmd = ['git', 'clean', '-fdx']
248    subprocess.run(cmd, cwd=proj_path, check=True)
249
250
251def is_valid_url(proj_path: Path, url: str) -> bool:
252    cmd = ['git', "ls-remote", url]
253    return subprocess.run(cmd, cwd=proj_path, check=False, stdin=subprocess.DEVNULL,
254                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
255                          start_new_session=True).returncode == 0
256
257
258def list_remote_tags(proj_path: Path, remote_name: str) -> list[str]:
259    """Lists tags in a remote repository."""
260    cmd = ['git', "ls-remote", "--tags", remote_name]
261    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
262                         text=True).stdout
263    lines = out.splitlines()
264    return lines
265
266
267def diff_stat(proj_path: Path, diff_filter: str, revision: str) -> str:
268    try:
269        cmd = ['git', 'diff', revision, '--stat', f'--diff-filter={diff_filter}']
270        out = subprocess.run(cmd, capture_output=True, cwd=proj_path,
271                             check=True, text=True).stdout
272        return out
273    except subprocess.CalledProcessError as err:
274        return f"Could not calculate the diff: {err}"
275
276
277def diff_name_only(proj_path: Path, diff_filter: str, revision: str) -> str:
278    try:
279        cmd = ['git', 'diff', revision, '--name-only', f'--diff-filter={diff_filter}']
280        out = subprocess.run(cmd, capture_output=True, cwd=proj_path,
281                             check=True, text=True).stdout
282        return out
283    except subprocess.CalledProcessError as err:
284        return f"Could not calculate the diff: {err}"
285
286
287def is_ancestor(proj_path: Path, ancestor: str, child: str) -> bool:
288    cmd = ['git', 'merge-base', '--is-ancestor', ancestor, child]
289    # https://git-scm.com/docs/git-merge-base#Documentation/git-merge-base.txt---is-ancestor
290    # Exit status of 0 means yes, 1 means no, and all others mean an error occurred.
291    # Although a commit is an ancestor of itself, we don't want to return True
292    # if ancestor points to the same commit as child.
293    if get_sha_for_branch(proj_path, ancestor) == child:
294        return False
295    try:
296        subprocess.run(
297            cmd,
298            cwd=proj_path,
299            text=True,
300            stderr=subprocess.STDOUT,
301            check=True,
302            stdout=subprocess.PIPE
303        )
304        return True
305    except subprocess.CalledProcessError as ex:
306        if ex.returncode == 1:
307            return False
308        raise
309
310
311def list_branches_with_commit(proj_path: Path, commit: str, remote_name: str) -> list[str]:
312    """Lists upstream branches which contain the specified commit"""
313    cmd = ['git', 'branch', '-r', '--contains', commit]
314    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
315                         text=True).stdout
316    lines = out.splitlines()
317    remote_branches = [line for line in lines if remote_name in line]
318    return remote_branches
319