1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
# 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
from typing import Dict, List, Tuple
import hashtags
import reviewers
def _run(cmd: List[str], cwd: Path) -> str:
"""Runs a command and returns its output."""
return subprocess.check_output(cmd, text=True, cwd=cwd)
def fetch(proj_path: Path, remote_names: List[str]) -> None:
"""Runs git fetch.
Args:
proj_path: Path to Git repository.
remote_names: Array of string to specify remote names.
"""
_run(['git', 'fetch', '--multiple'] + remote_names, cwd=proj_path)
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.
"""
_run(['git', 'remote', 'add', name, url], cwd=proj_path)
def remove_remote(proj_path: Path, name: str) -> None:
"""Removes a git remote."""
_run(['git', 'remote', 'remove', name], cwd=proj_path)
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])
out = _run(['git', 'remote', '-v'], proj_path)
lines = out.splitlines()
return dict([parse_remote(line) for line in lines])
def get_sha_for_branch(proj_path: Path, branch: str):
"""Gets the hash SHA for a branch."""
return _run(['git', 'rev-parse', branch], proj_path).strip()
def get_commits_ahead(proj_path: Path, branch: str,
base_branch: str) -> List[str]:
"""Lists commits in `branch` but not `base_branch`."""
out = _run([
'git', 'rev-list', '--left-only', '--ancestry-path', '{}...{}'.format(
branch, base_branch)
], proj_path)
return out.splitlines()
# pylint: disable=redefined-outer-name
def get_commit_time(proj_path: Path, commit: str) -> datetime.datetime:
"""Gets commit time of one commit."""
out = _run(['git', 'show', '-s', '--format=%ct', commit], cwd=proj_path)
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."""
lines = _run(['git', 'branch', '-r'], cwd=proj_path).splitlines()
stripped = [line.strip() for line in lines]
remote_path = remote_name + '/'
return [
line.lstrip(remote_path) for line in stripped
if line.startswith(remote_path)
]
def list_remote_tags(proj_path: Path, remote_name: str) -> List[str]:
"""Lists all tags for a remote."""
regex = re.compile(r".*refs/tags/(?P<tag>[^\^]*).*")
def parse_remote_tag(line: str) -> str:
return regex.match(line).group("tag")
lines = _run(['git', "ls-remote", "--tags", remote_name],
cwd=proj_path).splitlines()
tags = [parse_remote_tag(line) for line in lines]
return list(set(tags))
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:
_run(['git', 'merge', branch, '--no-commit'], cwd=proj_path)
except subprocess.CalledProcessError:
# Merge failed. Error is already written to console.
_run(['git', 'merge', '--abort'], cwd=proj_path)
raise
def add_file(proj_path: Path, file_name: str) -> None:
"""Stages a file."""
_run(['git', 'add', file_name], cwd=proj_path)
def delete_branch(proj_path: Path, branch_name: str) -> None:
"""Force delete a branch."""
_run(['git', 'branch', '-D', branch_name], cwd=proj_path)
def start_branch(proj_path: Path, branch_name: str) -> None:
"""Starts a new repo branch."""
_run(['repo', 'start', branch_name], cwd=proj_path)
def commit(proj_path: Path, message: str) -> None:
"""Commits changes."""
_run(['git', 'commit', '-m', message], cwd=proj_path)
def checkout(proj_path: Path, branch_name: str) -> None:
"""Checkouts a branch."""
_run(['git', 'checkout', branch_name], cwd=proj_path)
def push(proj_path: Path, remote_name: str) -> None:
"""Pushes change to remote."""
cmd = ['git', 'push', remote_name, 'HEAD:refs/for/master']
if revs := reviewers.find_reviewers(str(proj_path)):
cmd.extend(['-o', revs])
if tag := hashtags.find_hashtag(proj_path):
cmd.extend(['-o', 't=' + tag])
_run(cmd, cwd=proj_path)
|
