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
|
#!/usr/bin/env python3
#
# Copyright (c) 2019-2020 Red Hat, Inc.
#
# Author:
# Cleber Rosa <crosa@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2 or
# later. See the COPYING file in the top-level directory.
"""
Checks the GitLab pipeline status for a given commit ID
"""
# pylint: disable=C0103
import argparse
import http.client
import json
import os
import subprocess
import time
import sys
class CommunicationFailure(Exception):
"""Failed to communicate to gitlab.com APIs."""
class NoPipelineFound(Exception):
"""Communication is successfull but pipeline is not found."""
def get_local_branch_commit(branch):
"""
Returns the commit sha1 for the *local* branch named "staging"
"""
result = subprocess.run(['git', 'rev-parse', branch],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
cwd=os.path.dirname(__file__),
universal_newlines=True).stdout.strip()
if result == branch:
raise ValueError("There's no local branch named '%s'" % branch)
if len(result) != 40:
raise ValueError("Branch '%s' HEAD doesn't look like a sha1" % branch)
return result
def get_json_http_response(url):
"""
Returns the JSON content of an HTTP GET request to gitlab.com
"""
connection = http.client.HTTPSConnection('gitlab.com')
connection.request('GET', url=url)
response = connection.getresponse()
if response.code != http.HTTPStatus.OK:
msg = "Received unsuccessful response: %s (%s)" % (response.code,
response.reason)
raise CommunicationFailure(msg)
return json.loads(response.read())
def get_pipeline_status(project_id, commit_sha1):
"""
Returns the JSON content of the pipeline status API response
"""
url = '/api/v4/projects/{}/pipelines?sha={}'.format(project_id,
commit_sha1)
json_response = get_json_http_response(url)
# As far as I can tell, there should be only one pipeline for the same
# project + commit. If this assumption is false, we can add further
# filters to the url, such as username, and order_by.
if not json_response:
raise NoPipelineFound("No pipeline found")
return json_response[0]
def wait_on_pipeline_success(timeout, interval,
project_id, commit_sha):
"""
Waits for the pipeline to finish within the given timeout
"""
start = time.time()
while True:
if time.time() >= (start + timeout):
msg = ("Timeout (-t/--timeout) of %i seconds reached, "
"won't wait any longer for the pipeline to complete")
msg %= timeout
print(msg)
return False
try:
status = get_pipeline_status(project_id, commit_sha)
except NoPipelineFound:
print('Pipeline has not been found, it may not have been created yet.')
time.sleep(1)
continue
pipeline_status = status['status']
status_to_wait = ('created', 'waiting_for_resource', 'preparing',
'pending', 'running')
if pipeline_status in status_to_wait:
print('%s...' % pipeline_status)
time.sleep(interval)
continue
if pipeline_status == 'success':
return True
msg = "Pipeline failed, check: %s" % status['web_url']
print(msg)
return False
def create_parser():
parser = argparse.ArgumentParser(
prog='pipeline-status',
description='check or wait on a pipeline status')
parser.add_argument('-t', '--timeout', type=int, default=7200,
help=('Amount of time (in seconds) to wait for the '
'pipeline to complete. Defaults to '
'%(default)s'))
parser.add_argument('-i', '--interval', type=int, default=60,
help=('Amount of time (in seconds) to wait between '
'checks of the pipeline status. Defaults '
'to %(default)s'))
parser.add_argument('-w', '--wait', action='store_true', default=False,
help=('Wether to wait, instead of checking only once '
'the status of a pipeline'))
parser.add_argument('-p', '--project-id', type=int, default=11167699,
help=('The GitLab project ID. Defaults to the project '
'for https://gitlab.com/qemu-project/qemu, that '
'is, "%(default)s"'))
parser.add_argument('-b', '--branch', type=str, default="staging",
help=('Specify the branch to check. '
'Use HEAD for your current branch. '
'Otherwise looks at "%(default)s"'))
parser.add_argument('-c', '--commit',
default=None,
help=('Look for a pipeline associated with the given '
'commit. If one is not explicitly given, the '
'commit associated with the default branch '
'is used.'))
parser.add_argument('--verbose', action='store_true', default=False,
help=('A minimal verbosity level that prints the '
'overall result of the check/wait'))
return parser
def main():
"""
Script entry point
"""
parser = create_parser()
args = parser.parse_args()
if not args.commit:
args.commit = get_local_branch_commit(args.branch)
success = False
try:
if args.wait:
success = wait_on_pipeline_success(
args.timeout,
args.interval,
args.project_id,
args.commit)
else:
status = get_pipeline_status(args.project_id,
args.commit)
success = status['status'] == 'success'
except Exception as error: # pylint: disable=W0703
if args.verbose:
print("ERROR: %s" % error.args[0])
except KeyboardInterrupt:
if args.verbose:
print("Exiting on user's request")
if success:
if args.verbose:
print('success')
sys.exit(0)
else:
if args.verbose:
print('failure')
sys.exit(1)
if __name__ == '__main__':
main()
|
