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
|
#!/usr/bin/env python3
import click
import discord_webhook
import multiprocessing
import os
import pathlib
import requests
import urllib.parse
@click.command()
@click.option('--token_path', type=str, default=pathlib.Path.home() / '.gitlab_token',
help='Path to text file with Gitlab token.')
@click.option('--project_id', type=int, default=7107382,
help='Gitlab project id.')
@click.option('--job_id', type=int, default=os.getenv('CI_JOB_ID'),
help='Gitlab job id.')
@click.option('--host', type=str, default='gitlab.com',
help='Gitlab host.')
@click.option('--workers', type=int, default=10,
help='Number of parallel workers.')
@click.option('--target_branch', type=str, default='master',
help='Merge request target branch.')
@click.option('--begin_page', type=int, default=1,
help='Begin with given /merge_requests page.')
@click.option('--end_page', type=int, default=4,
help='End before given /merge_requests page.')
@click.option('--per_page', type=int, default=100,
help='Number of merge requests per page.')
@click.option('--ignored_mrs_path', type=str,
help='Path to a list of ignored MRs.')
def main(token_path, project_id, job_id, host, workers, target_branch, begin_page, end_page, per_page, ignored_mrs_path):
headers = make_headers(token_path)
base_url = f'https://{host}/api/v4/projects/{project_id}/'
discord_webhook_url = os.getenv('DISCORD_WEBHOOK_URL')
ignored_mrs = frozenset(read_ignored_mrs(ignored_mrs_path))
checked = 0
filtered = 0
missing = 0
missing_mrs = list()
for page in range(begin_page, end_page):
merge_requests = parse_gitlab_response(requests.get(
url=urllib.parse.urljoin(base_url, 'merge_requests'),
headers=headers,
params=dict(state='merged', per_page=per_page, page=page),
))
if not merge_requests:
break
checked += len(merge_requests)
merge_requests = [v for v in merge_requests if v['target_branch'] == target_branch]
if not merge_requests:
continue
filtered += len(merge_requests)
with multiprocessing.Pool(workers) as pool:
missing_merge_requests = pool.map(FilterMissingMergeRequest(headers, base_url), merge_requests)
for mr in missing_merge_requests:
if mr is None:
continue
if mr['reference'] in ignored_mrs or mr['reference'].strip('!') in ignored_mrs:
print(f"Ignored MR {mr['reference']} ({mr['id']}) is missing from branch {mr['target_branch']},"
f" previously was merged as {mr['merge_commit_sha']}")
continue
missing += 1
missing_mrs.append(mr)
print(f"MR {mr['reference']} ({mr['id']}) is missing from branch {mr['target_branch']},"
f" previously was merged as {mr['merge_commit_sha']}")
print(f'Checked {checked} MRs ({filtered} with {target_branch} target branch), {missing} are missing')
if discord_webhook_url is not None and missing_mrs:
project_web_url = parse_gitlab_response(requests.get(url=base_url, headers=headers))['web_url'] + '/'
discord_message = format_discord_message(missing=missing, filtered=filtered, target_branch=target_branch,
project_web_url=project_web_url, missing_mrs=missing_mrs,
job_id=job_id)
print('Sending Discord notification...')
print(discord_message)
discord_webhook.DiscordWebhook(url=discord_webhook_url, content=discord_message, rate_limit_retry=True).execute()
if missing_mrs:
exit(-1)
def format_discord_message(missing, filtered, target_branch, project_web_url, missing_mrs, job_id):
target_branch = format_link(target_branch, urllib.parse.urljoin(project_web_url, f'-/tree/{target_branch}'))
job = f' by job ' + format_link(job_id, urllib.parse.urljoin(project_web_url, f'-/jobs/{job_id}')) if job_id else ''
return (
f'Found {missing} missing MRs out of {filtered} from {target_branch} target branch{job}:\n'
+ '\n'.join(format_missing_mr_message(v, project_web_url) for v in missing_mrs)
)
def format_missing_mr_message(mr, project_web_url):
web_url = mr.get('web_url')
reference = mr['reference']
target_branch = mr['target_branch']
commit = mr['merge_commit_sha']
if web_url is not None:
reference = format_link(reference, web_url)
target_branch = format_link(target_branch, urllib.parse.urljoin(project_web_url, f'-/tree/{target_branch}'))
commit = format_link(commit, urllib.parse.urljoin(project_web_url, f'-/commit/{commit}'))
return f"MR {reference} is missing from branch {target_branch}, previously was merged as {commit}"
def format_link(name, url):
return f'[{name}]({url})'
class FilterMissingMergeRequest:
def __init__(self, headers, base_url):
self.headers = headers
self.base_url = base_url
def __call__(self, merge_request):
commit_refs = requests.get(
url=urllib.parse.urljoin(self.base_url, f"repository/commits/{merge_request['merge_commit_sha']}/refs"),
headers=self.headers,
).json()
if 'message' in commit_refs and commit_refs['message'] == '404 Commit Not Found':
return merge_request
if not present_in_branch(commit_refs, branch=merge_request['target_branch']):
return merge_request
def present_in_branch(commit_refs, branch):
return bool(next((v for v in commit_refs if v['type'] == 'branch' and v['name'] == branch), None))
def make_headers(token_path):
job_token = os.environ.get('CI_JOB_TOKEN')
if job_token is not None:
print('Using auth token from CI_JOB_TOKEN env')
return {'JOB_TOKEN': job_token}
if not os.path.exists(token_path):
print(f'Ignore absent token path: {token_path}')
return dict()
print(f'Using auth token from: {token_path}')
return {'PRIVATE-TOKEN': read_token(token_path)}
def read_ignored_mrs(path):
if path is None:
return
with open(path) as stream:
for line in stream:
yield line.strip()
def read_token(path):
with open(path) as stream:
return stream.readline().strip()
def parse_gitlab_response(response):
response = response.json()
if isinstance(response, dict):
message = response.get('message')
if message is not None:
raise RuntimeError(f'Gitlab request has failed: {message}')
return response
if __name__ == '__main__':
main()
|
