• 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 shutil
19import subprocess
20from pathlib import Path
21
22import hashtags
23import reviewers
24
25
26def fetch(proj_path: Path, remote_names: list[str]) -> None:
27    """Runs git fetch.
28
29    Args:
30        proj_path: Path to Git repository.
31        remote_names: Array of string to specify remote names.
32    """
33    cmd = ['git', 'fetch', '--tags', '--multiple'] + remote_names
34    subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True)
35
36
37def add_remote(proj_path: Path, name: str, url: str) -> None:
38    """Adds a git remote.
39
40    Args:
41        proj_path: Path to Git repository.
42        name: Name of the new remote.
43        url: Url of the new remote.
44    """
45    cmd = ['git', 'remote', 'add', name, url]
46    subprocess.run(cmd, cwd=proj_path, check=True)
47
48
49def remove_remote(proj_path: Path, name: str) -> None:
50    """Removes a git remote."""
51    cmd = ['git', 'remote', 'remove', name]
52    subprocess.run(cmd, cwd=proj_path, check=True)
53
54
55def list_remotes(proj_path: Path) -> dict[str, str]:
56    """Lists all Git remotes.
57
58    Args:
59        proj_path: Path to Git repository.
60
61    Returns:
62        A dict from remote name to remote url.
63    """
64    def parse_remote(line: str) -> tuple[str, str]:
65        split = line.split()
66        return split[0], split[1]
67
68    cmd = ['git', 'remote', '-v']
69    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
70                         text=True).stdout
71    lines = out.splitlines()
72    return dict([parse_remote(line) for line in lines])
73
74
75def detect_default_branch(proj_path: Path, remote_name: str) -> str:
76    """Gets the name of the upstream's default branch to use."""
77    cmd = ['git', 'remote', 'show', remote_name]
78    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
79                         text=True).stdout
80    lines = out.splitlines()
81    for line in lines:
82        if "HEAD branch" in line:
83            return line.split()[-1]
84    raise RuntimeError(
85        f"Could not find HEAD branch in 'git remote show {remote_name}'"
86    )
87
88
89def get_sha_for_branch(proj_path: Path, branch: str):
90    """Gets the hash SHA for a branch."""
91    cmd = ['git', 'rev-parse', branch]
92    return subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
93                          text=True).stdout.strip()
94
95
96def get_commits_ahead(proj_path: Path, branch: str,
97                      base_branch: str) -> list[str]:
98    """Lists commits in `branch` but not `base_branch`."""
99    cmd = [
100        'git', 'rev-list', '--left-only', '--ancestry-path', 'f{branch}...{base_branch}'
101    ]
102    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
103                         text=True).stdout
104    return out.splitlines()
105
106
107# pylint: disable=redefined-outer-name
108def get_commit_time(proj_path: Path, commit: str) -> datetime.datetime:
109    """Gets commit time of one commit."""
110    cmd = ['git', 'show', '-s', '--format=%ct', commit]
111    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
112                         text=True).stdout
113    return datetime.datetime.fromtimestamp(int(out.strip()))
114
115
116def list_remote_branches(proj_path: Path, remote_name: str) -> list[str]:
117    """Lists all branches for a remote."""
118    cmd = ['git', 'branch', '-r']
119    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
120                           text=True).stdout.splitlines()
121    stripped = [line.strip() for line in lines]
122    remote_path = remote_name + '/'
123    return [
124        line[len(remote_path):] for line in stripped
125        if line.startswith(remote_path)
126    ]
127
128
129def list_local_branches(proj_path: Path) -> list[str]:
130    """Lists all local branches."""
131    cmd = ['git', 'branch', '--format=%(refname:short)']
132    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
133                           text=True).stdout.splitlines()
134    return lines
135
136
137def list_remote_tags(proj_path: Path, remote_name: str) -> list[str]:
138    """Lists all tags for a remote."""
139    regex = re.compile(r".*refs/tags/(?P<tag>[^\^]*).*")
140
141    def parse_remote_tag(line: str) -> str:
142        if (m := regex.match(line)) is not None:
143            return m.group("tag")
144        raise ValueError(f"Could not parse tag from {line}")
145
146    cmd = ['git', "ls-remote", "--tags", remote_name]
147    lines = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
148                           text=True).stdout.splitlines()
149    tags = [parse_remote_tag(line) for line in lines]
150    return list(set(tags))
151
152
153COMMIT_PATTERN = r'^[a-f0-9]{40}$'
154COMMIT_RE = re.compile(COMMIT_PATTERN)
155
156
157# pylint: disable=redefined-outer-name
158def is_commit(commit: str) -> bool:
159    """Whether a string looks like a SHA1 hash."""
160    return bool(COMMIT_RE.match(commit))
161
162
163def merge(proj_path: Path, branch: str) -> None:
164    """Merges a branch."""
165    try:
166        cmd = ['git', 'merge', branch, '--no-commit']
167        subprocess.run(cmd, cwd=proj_path, check=True)
168    except subprocess.CalledProcessError as err:
169        if hasattr(err, "output"):
170            print(err.output)
171        if not merge_conflict(proj_path):
172            raise
173
174
175def merge_conflict(proj_path: Path) -> bool:
176    """Checks if there was a merge conflict."""
177    cmd = ['git', 'ls-files', '--unmerged']
178    out = subprocess.run(cmd, capture_output=True, cwd=proj_path, check=True,
179                         text=True).stdout
180    return bool(out)
181
182
183def add_file(proj_path: Path, file_name: str) -> None:
184    """Stages a file."""
185    cmd = ['git', 'add', file_name]
186    subprocess.run(cmd, cwd=proj_path, check=True)
187
188
189def remove_gitmodules(proj_path: Path) -> None:
190    """Deletes .gitmodules files."""
191    cmd = ['find', '.', '-name', '.gitmodules', '-delete']
192    subprocess.run(cmd, cwd=proj_path, check=True)
193
194
195def delete_branch(proj_path: Path, branch_name: str) -> None:
196    """Force delete a branch."""
197    cmd = ['git', 'branch', '-D', branch_name]
198    subprocess.run(cmd, cwd=proj_path, check=True)
199
200
201def tree_uses_pore(proj_path: Path) -> bool:
202    """Returns True if the tree uses pore rather than repo.
203
204    https://github.com/jmgao/pore
205    """
206    if shutil.which("pore") is None:
207        # Fast path for users that don't have pore installed, since that's almost
208        # everyone.
209        return False
210
211    if proj_path == Path(proj_path.root):
212        return False
213    if (proj_path / ".pore").exists():
214        return True
215    return tree_uses_pore(proj_path.parent)
216
217
218def start_branch(proj_path: Path, branch_name: str) -> None:
219    """Starts a new repo branch."""
220    repo = 'repo'
221    if tree_uses_pore(proj_path):
222        repo = 'pore'
223    cmd = [repo, 'start', branch_name]
224    subprocess.run(cmd, cwd=proj_path, check=True)
225
226
227def commit(proj_path: Path, message: str) -> None:
228    """Commits changes."""
229    cmd = ['git', 'commit', '-m', message]
230    subprocess.run(cmd, cwd=proj_path, check=True)
231
232
233def checkout(proj_path: Path, branch_name: str) -> None:
234    """Checkouts a branch."""
235    cmd = ['git', 'checkout', branch_name]
236    subprocess.run(cmd, cwd=proj_path, check=True)
237
238
239def push(proj_path: Path, remote_name: str, has_errors: bool) -> None:
240    """Pushes change to remote."""
241    cmd = ['git', 'push', remote_name, 'HEAD:refs/for/master']
242    if revs := reviewers.find_reviewers(str(proj_path)):
243        cmd.extend(['-o', revs])
244    if tag := hashtags.find_hashtag(proj_path):
245        cmd.extend(['-o', 't=' + tag])
246    if has_errors:
247        cmd.extend(['-o', 'l=Verified-1'])
248    subprocess.run(cmd, cwd=proj_path, check=True)
249
250
251def reset_hard(proj_path: Path) -> None:
252    """Resets current HEAD and discards changes to tracked files."""
253    cmd = ['git', 'reset', '--hard']
254    subprocess.run(cmd, cwd=proj_path, check=True)
255
256
257def clean(proj_path: Path) -> None:
258    """Removes untracked files and directories."""
259    cmd = ['git', 'clean', '-fdx']
260    subprocess.run(cmd, cwd=proj_path, check=True)
261
262
263def is_valid_url(proj_path: Path, url: str) -> bool:
264    cmd = ['git', "ls-remote", url]
265    return subprocess.run(cmd, cwd=proj_path, stdin=subprocess.DEVNULL,
266                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
267                          start_new_session=True).returncode == 0
268