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
|
#!/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
def get_local_branch_commit(branch='staging'):
"""
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_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)
connection = http.client.HTTPSConnection('gitlab.com')
connection.request('GET', url=url)
response = connection.getresponse()
if response.code != http.HTTPStatus.OK:
raise ValueError("Failed to receive a successful response")
json_response = json.loads(response.read())
# 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 ValueError("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
status = get_pipeline_status(project_id, commit_sha)
if status['status'] == 'running':
print('running...')
time.sleep(interval)
continue
if status['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"'))
try:
default_commit = get_local_branch_commit()
commit_required = False
except ValueError:
default_commit = ''
commit_required = True
parser.add_argument('-c', '--commit', required=commit_required,
default=default_commit,
help=('Look for a pipeline associated with the given '
'commit. If one is not explicitly given, the '
'commit associated with the local branch named '
'"staging" is used. Default: %(default)s'))
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()
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
success = False
if args.verbose:
print("ERROR: %s" % error.args[0])
if success:
if args.verbose:
print('success')
sys.exit(0)
else:
if args.verbose:
print('failure')
sys.exit(1)
if __name__ == '__main__':
main()
|
