#!/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_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 CommunicationFailure("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 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()
