/
active_branch.py
163 lines (125 loc) · 5.49 KB
/
active_branch.py
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
import re
import string
class ActiveBranchMixin():
def get_current_branch_name(self):
"""
Return the name of the last checkout-out branch.
"""
stdout = self.git("branch", "--no-color")
try:
correct_line = next(line for line in stdout.split("\n") if line.startswith("*"))
return correct_line[2:]
except StopIteration:
return None
def _get_branch_status_components(self):
"""
Return a tuple of:
0) boolean indicating whether repo is in detached state
1) boolean indicating whether this is initial commit
2) active branch name
3) remote branch name
4) boolean indicating whether branch is clean
5) # commits ahead of remote
6) # commits behind of remote
7) boolean indicating whether the remote branch is gone
"""
stdout = self.git("status", "-b", "--porcelain").strip()
first_line, *addl_lines = stdout.split("\n", 2)
# Any additional lines will mean files have changed or are untracked.
clean = len(addl_lines) == 0
if first_line.startswith("## HEAD (no branch)"):
return True, False, None, None, clean, None, None, False
if first_line.startswith("## Initial commit on "):
return False, True, first_line[21:], clean, None, None, None, False
valid_punctuation = "".join(c for c in string.punctuation if c not in "~^:?*[\\")
branch_pattern = "[A-Za-z0-9" + re.escape(valid_punctuation) + "\u263a-\U0001f645]+?"
branch_suffix = r"( \[((ahead (\d+))(, )?)?(behind (\d+))?(gone)?\])?)"
short_status_pattern = "## (" + branch_pattern + r")(\.\.\.(" + branch_pattern + ")" + branch_suffix + "?$"
status_match = re.match(short_status_pattern, first_line)
if not status_match:
return False, False, None if clean else addl_lines[0], None, clean, None, None, False
branch, _, remote, _, _, _, ahead, _, _, behind, gone = status_match.groups()
return False, False, branch, remote, clean, ahead, behind, bool(gone)
def get_branch_status(self, delim=None):
"""
Return a tuple of:
1) the name of the active branch
2) the status of the active local branch
compared to its remote counterpart.
If no remote or tracking branch is defined, do not include remote-data.
If HEAD is detached, provide that status instead.
If a delimeter is provided, join tuple components with it, and return
that value.
"""
detached, initial, branch, remote, clean, ahead, behind, gone = \
self._get_branch_status_components()
secondary = []
if detached:
status = "HEAD is in a detached state."
elif initial:
status = "Initial commit on `{}`.".format(branch)
else:
tracking = " tracking `{}`".format(remote)
status = "On branch `{}`{}.".format(branch, tracking if remote else "")
if ahead and behind:
secondary.append("You're ahead by {} and behind by {}.".format(ahead, behind))
elif ahead:
secondary.append("You're ahead by {}.".format(ahead))
elif behind:
secondary.append("You're behind by {}.".format(behind))
elif gone:
secondary.append("The remote branch is gone.")
if self.in_merge():
secondary.append("Merging {}.".format(self.merge_head()))
if self.in_rebase():
secondary.append("Rebasing {}.".format(self.rebase_branch_name()))
if delim:
return delim.join([status] + secondary) if secondary else status
return status, secondary
def get_branch_status_short(self):
if self.in_rebase():
return "(no branch, rebasing {})".format(self.rebase_branch_name())
merge_head = self.merge_head() if self.in_merge() else ""
detached, initial, branch, remote, clean, ahead, behind, gone = \
self._get_branch_status_components()
dirty = "" if clean else "*"
if detached:
return "DETACHED" + dirty
output = branch + dirty
if ahead:
output += "+" + ahead
if behind:
output += "-" + behind
return output if not merge_head else output + " (merging {})".format(merge_head)
def get_commit_hash_for_head(self):
"""
Get the SHA1 commit hash for the commit at HEAD.
"""
return self.git("rev-parse", "HEAD").strip()
def get_latest_commit_msg_for_head(self):
"""
Get last commit msg for the commit at HEAD.
"""
stdout = self.git(
"log",
"-n 1",
"--pretty=format:%h %s",
"--abbrev-commit",
throw_on_stderr=False
).strip()
return stdout or "No commits yet."
def get_upstream_for_active_branch(self):
"""
Return ref for remote tracking branch.
"""
return self.git("rev-parse", "--abbrev-ref", "--symbolic-full-name",
"@{u}", throw_on_stderr=False).strip()
def get_active_remote_branch(self):
"""
Return named tuple of the upstream for active branch.
"""
upstream = self.get_upstream_for_active_branch()
for branch in self.get_branches():
if branch.name_with_remote == upstream:
return branch
return None