-
Notifications
You must be signed in to change notification settings - Fork 210
/
prs.py
291 lines (225 loc) · 10.1 KB
/
prs.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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
import arrow
import math
import settings
from . import misc
from . import voting
from . import comments
from . import commits
from . import exceptions as exc
def merge_pr(api, urn, pr, votes, total, threshold):
""" merge a pull request, if possible, and use a nice detailed merge commit
message """
pr_num = pr["number"]
pr_title = pr['title']
pr_description = pr['body']
path = "/repos/{urn}/pulls/{pr}/merge".format(urn=urn, pr=pr_num)
record = voting.friendly_voting_record(votes)
if record:
record = "Vote record:\n" + record
votes_summary = formatted_votes_summary(votes, total, threshold)
pr_url = "https://github.com/{urn}/pull/{pr}".format(urn=urn, pr=pr_num)
title = "merging PR #{num}: {pr_title}".format(
num=pr_num, pr_title=pr_title)
desc = """
{pr_url}: {pr_title}
Description:
{pr_description}
:ok_woman: PR passed {summary}.
{record}
""".strip().format(
pr_url=pr_url,
pr_title=pr_title,
pr_description=pr_description,
summary=votes_summary,
record=record,
)
data = {
"commit_title": title,
"commit_message": desc,
"merge_method": "merge",
# if some clever person attempts to submit more commits while we're
# aggregating votes, this sha check will fail and no merge will occur
"sha": pr["head"]["sha"],
}
try:
resp = api("PUT", path, json=data)
return resp["sha"]
except HTTPError as e:
resp = e.response
# could not be merged
if resp.status_code == 405:
raise exc.CouldntMerge
# someone trying to be sneaky and change their PR commits during voting
elif resp.status_code == 409:
raise exc.CouldntMerge
else:
raise
def formatted_votes_summary(votes, total, threshold):
vfor = sum(v for v in votes.values() if v > 0)
vagainst = abs(sum(v for v in votes.values() if v < 0))
return "with a vote of {vfor} for and {vagainst} against, with a weighted total of {total:.1f} and a threshold of {threshold:.1f}" \
.strip().format(vfor=vfor, vagainst=vagainst, total=total, threshold=threshold)
def formatted_votes_short_summary(votes, total, threshold):
vfor = sum(v for v in votes.values() if v > 0)
vagainst = abs(sum(v for v in votes.values() if v < 0))
return "vote: {vfor}-{vagainst}, weighted total: {total:.1f}, threshold: {threshold:.1f}" \
.strip().format(vfor=vfor, vagainst=vagainst, total=total, threshold=threshold)
def label_pr(api, urn, pr_num, labels):
""" set a pr's labels (removes old labels) """
if not isinstance(labels, (tuple, list)):
labels = [labels]
path = "/repos/{urn}/issues/{pr}/labels".format(urn=urn, pr=pr_num)
data = labels
resp = api("PUT", path, json=data)
def close_pr(api, urn, pr):
""" https://developer.github.com/v3/pulls/#update-a-pull-request """
path = "/repos/{urn}/pulls/{pr}".format(urn=urn, pr=pr["number"])
data = {
"state": "closed",
}
return api("patch", path, json=data)
def get_pr_last_updated(api, urn, pr_data):
""" a helper for finding the utc datetime of the last pr branch
modifications """
# read about github's merge test commit
# https://developer.github.com/v3/pulls/#get-a-single-pull-request
# essentially what github does is, when it sees a new push to a PR, it will
# start a background job to test the PR for mergeability. this
# background job creates a magical floating merge commit not really attached
# to any branch. however, this commit is real and its committer is
# "GitHub"...and it has a commit date that we can assume is never malicious.
# And it's always updated on a new push to the PR's branch
#
# using all of these facts, we are able to determine the true, reliable,
# last update time of the branch backing a PR
#
# NOTE could potentially be deprecated at some point
# https://developer.github.com/changes/2013-04-25-deprecating-merge-commit-sha/
#
# based on https://stackoverflow.com/q/37442144/345059, it looks like it can
# be an empty string if the background job is still running, but also handle
# cases where the key doesn't exist, and also normalize to None if empty
github_test_merge_commit = pr_data.get("merge_commit_sha", None) or None
updated = None
if github_test_merge_commit:
commit = commits.get_commit(api, urn, github_test_merge_commit)
updated = arrow.get(commit["commit"]["committer"]["date"])
return updated
def get_pr_comments(api, urn, pr_num):
""" yield all comments on a pr, weirdly excluding the initial pr comment
itself (the one the owner makes) """
params = {
"per_page": settings.DEFAULT_PAGINATION
}
path = "/repos/{urn}/issues/{pr}/comments".format(urn=urn, pr=pr_num)
comments = api("get", path, params=params)
for comment in comments:
yield comment
def get_ready_prs(api, urn, window):
""" yield mergeable, non-WIP prs that have had no modifications for longer
than the voting window. these are prs that are ready to be considered for
merging """
open_prs = get_open_prs(api, urn)
for pr in open_prs:
pr_num = pr["number"]
updated = get_pr_last_updated(api, urn, pr)
# if there's no updated time, don't even consider this PR
if not updated:
continue
now = arrow.utcnow()
delta = (now - updated).total_seconds()
is_wip = "WIP" in pr["title"]
if not is_wip and delta > window:
# we check if its mergeable if its outside the voting window,
# because there seems to be a race where a freshly-created PR exists
# in the paginated list of PRs, but 404s when trying to fetch it
# directly
mergeable = get_is_mergeable(api, urn, pr_num)
if mergeable is True:
label_pr(api, urn, pr_num, [])
yield pr
elif mergeable is False:
label_pr(api, urn, pr_num, ["conflicts"])
if delta >= 60 * 60 * settings.PR_STALE_HOURS:
comments.leave_stale_comment(
api, urn, pr["number"], round(delta / 60 / 60))
close_pr(api, urn, pr)
# mergeable can also be None, in which case we just skip it for now
def voting_window_remaining_seconds(api, urn, pr, window):
""" returns the number of seconds until voting is over. can be negative,
meaning voting has been over for that long """
now = arrow.utcnow()
pr_updated = get_pr_last_updated(api, urn, pr)
# this is how many seconds ago the pr has been updated with new commits.
# if we don't have a last update time, we're setting this to negative
# infinity, which is a mind-bender, but makes the maths work out
elapsed_last_update = -math.inf
if pr_updated:
elapsed_last_update = (now - pr_updated).total_seconds()
return window - elapsed_last_update
def is_pr_in_voting_window(api, urn, pr, window):
return voting_window_remaining_seconds(api, urn, pr, window) <= 0
def get_pr_reviews(api, urn, pr_num):
""" get all pr reviews on a pr
https://help.github.com/articles/about-pull-request-reviews/ """
params = {
"per_page": settings.DEFAULT_PAGINATION
}
path = "/repos/{urn}/pulls/{pr}/reviews".format(urn=urn, pr=pr_num)
data = api("get", path, params=params)
return data
def get_is_mergeable(api, urn, pr_num):
return get_pr(api, urn, pr_num)["mergeable"]
def get_pr(api, urn, pr_num):
""" helper for fetching a pr. necessary because the "mergeable" field does
not exist on prs that come back from paginated endpoints, so we must fetch
the pr directly """
path = "/repos/{urn}/pulls/{pr}".format(urn=urn, pr=pr_num)
pr = api("get", path)
return pr
def get_open_prs(api, urn):
params = {
"state": "open",
"sort": "updated",
"direction": "asc",
"per_page": settings.DEFAULT_PAGINATION,
}
path = "/repos/{urn}/pulls".format(urn=urn)
data = api("get", path, params=params)
return data
def get_reactions_for_pr(api, urn, pr):
path = "/repos/{urn}/issues/{pr}/reactions".format(urn=urn, pr=pr)
params = {"per_page": settings.DEFAULT_PAGINATION}
reactions = api("get", path, params=params)
for reaction in reactions:
yield reaction
def post_accepted_status(api, urn, pr, voting_window, votes, total, threshold):
sha = pr["head"]["sha"]
remaining_seconds = voting_window_remaining_seconds(api, urn, pr, voting_window)
remaining_human = misc.seconds_to_human(remaining_seconds)
votes_summary = formatted_votes_short_summary(votes, total, threshold)
post_status(api, urn, sha, "success",
"remaining: {time}, {summary}".format(time=remaining_human, summary=votes_summary))
def post_rejected_status(api, urn, pr, voting_window, votes, total, threshold):
sha = pr["head"]["sha"]
remaining_seconds = voting_window_remaining_seconds(api, urn, pr, voting_window)
remaining_human = misc.seconds_to_human(remaining_seconds)
votes_summary = formatted_votes_short_summary(votes, total, threshold)
post_status(api, urn, sha, "failure",
"remaining: {time}, {summary}".format(time=remaining_human, summary=votes_summary))
def post_pending_status(api, urn, pr, voting_window, votes, total, threshold):
sha = pr["head"]["sha"]
remaining_seconds = voting_window_remaining_seconds(api, urn, pr, voting_window)
remaining_human = misc.seconds_to_human(remaining_seconds)
votes_summary = formatted_votes_short_summary(votes, total, threshold)
post_status(api, urn, sha, "pending",
"remaining: {time}, {summary}".format(time=remaining_human, summary=votes_summary))
def post_status(api, urn, sha, state, description):
""" apply an issue label to a pr """
path = "/repos/{urn}/statuses/{sha}".format(urn=urn, sha=sha)
data = {
"state": state,
"description": description,
"context": "chaosbot"
}
api("POST", path, json=data)