419 lines
13 KiB
Python
419 lines
13 KiB
Python
#!/usr/bin/python
|
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
# Error Codes
|
|
# 1000: Failed check (default error)
|
|
# 1001: Check not existing but expected
|
|
|
|
import argparse
|
|
import logging
|
|
import requests
|
|
import json
|
|
import re
|
|
|
|
|
|
class FailedPR:
|
|
"""Base class for failed Pull Requests"""
|
|
|
|
def __init__(
|
|
self,
|
|
created_at,
|
|
host,
|
|
updated_at,
|
|
url,
|
|
error=None,
|
|
pullrequest=None,
|
|
org=None,
|
|
repo=None,
|
|
status=None,
|
|
zuul_url=None):
|
|
|
|
self.created_at = created_at
|
|
self.error = error
|
|
self.host = host
|
|
self.org = org
|
|
self.pullrequest = pullrequest
|
|
self.repo = repo
|
|
self.status = status
|
|
self.updated_at = updated_at
|
|
self.url = url
|
|
self.zuul_url = zuul_url
|
|
|
|
|
|
def get_args():
|
|
"""
|
|
Function to collect parameter.
|
|
"""
|
|
parser = argparse.ArgumentParser(
|
|
description='Bootstrap repositories.')
|
|
parser.add_argument(
|
|
'--gh-orgs',
|
|
nargs='+',
|
|
default=['opentelekomcloud-docs'],
|
|
help='One or more GitHub organizations to be queried for failed Pull '
|
|
'Requests.\n'
|
|
'Default: [opentelekomcloud-docs]'
|
|
)
|
|
parser.add_argument(
|
|
'--gitea-orgs',
|
|
nargs='+',
|
|
default=['docs'],
|
|
help='One or more Gitea organizations to be queried for failed Pull '
|
|
'Requests.\n'
|
|
'Default: [docs]'
|
|
)
|
|
parser.add_argument(
|
|
'--gh-token',
|
|
help='API Token for GitHub.'
|
|
)
|
|
parser.add_argument(
|
|
'--gitea-token',
|
|
help='API Token for Gitea'
|
|
)
|
|
parser.add_argument(
|
|
'--debug',
|
|
action='store_true',
|
|
help='Set debug mode on.'
|
|
)
|
|
parser.add_argument(
|
|
'--gitea-url',
|
|
default='https://gitea.eco.tsi-dev.otc-service.com/api/v1/',
|
|
help='Gitea base URL for API request.\n'
|
|
'Default: https://gitea.eco.tsi-dev.otc-service.com/api/v1/'
|
|
)
|
|
parser.add_argument(
|
|
'--gh-url',
|
|
default='https://api.github.com/',
|
|
help='GitHub Base URL for API request.\n'
|
|
'Default: https://api.github.com/'
|
|
)
|
|
parser.add_argument(
|
|
'--hoster',
|
|
default=['gitea', 'github'],
|
|
nargs='+',
|
|
help='Git hoster to be queried for failed Pull Requests.\n'
|
|
'Default: [github, gitea].'
|
|
)
|
|
return parser.parse_args()
|
|
|
|
def add_builds_to_obj(obj, url, tenant):
|
|
"""
|
|
This method trys to find all build jobs under a Zuul buildset.
|
|
The corresponding data like log_url and status will be added.
|
|
"""
|
|
zuul_api_url = "https://zuul.otc-service.com/api/tenant/" + tenant + "/buildset/"
|
|
final_url = re.sub('.*\/buildset\/', zuul_api_url, url)
|
|
zuul_headers = {}
|
|
zuul_headers['accept'] = 'application/json'
|
|
res_zuul = requests.request('GET', url=final_url, headers=zuul_headers)
|
|
if res_zuul.status_code != 404 and res_zuul.json():
|
|
x = res_zuul.json()
|
|
if len(x['builds']) != 0:
|
|
jobs = []
|
|
for build in x['builds']:
|
|
job = {}
|
|
job['uuid'] = build['uuid']
|
|
job['name'] = build['job_name']
|
|
job['result'] = build['result']
|
|
job['log_url'] = build['log_url']
|
|
jobs.append(job)
|
|
obj.jobs = jobs
|
|
return obj
|
|
|
|
def get_gitea_repos(url, headers, gitea_org):
|
|
"""
|
|
Get all Repositories of one Gitea orgainzation
|
|
"""
|
|
repositories = []
|
|
i = 1
|
|
|
|
while True:
|
|
try:
|
|
req_url = (url
|
|
+ 'orgs/'
|
|
+ gitea_org
|
|
+ '/repos?limit=50&page='
|
|
+ str(i))
|
|
res = requests.request('GET', url=req_url, headers=headers)
|
|
i += 1
|
|
if res.json():
|
|
for repo in res.json():
|
|
repositories.append(repo)
|
|
continue
|
|
else:
|
|
break
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res.status_code) +
|
|
" | " + str(res.reason))
|
|
break
|
|
return repositories
|
|
|
|
def get_gitea_prs(url, headers, gitea_org, repo):
|
|
"""
|
|
Collect all Pull Requests of a Gitea Repository
|
|
"""
|
|
pullrequests = []
|
|
|
|
try:
|
|
req_url = (url + 'repos/' + gitea_org + '/'
|
|
+ repo + '/pulls?state=open')
|
|
res = requests.request('GET', url=req_url, headers=headers)
|
|
if res.json():
|
|
for pr in res.json():
|
|
pullrequests.append(pr)
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res.status_code) +
|
|
" | " + str(res.reason))
|
|
exit()
|
|
return pullrequests
|
|
|
|
def get_failed_gitea_commits(pull, url, gitea_org, repo, headers):
|
|
"""
|
|
Collect all failed gitea commits of one repository.
|
|
"""
|
|
failed_commits = []
|
|
try:
|
|
req_url = (url + 'repos/' + gitea_org + '/' + repo['name'] +
|
|
'/commits/' + pull['head']['ref'] + '/statuses?limit=1')
|
|
res_sta = requests.request('GET', url=req_url, headers=headers)
|
|
if res_sta.json():
|
|
if res_sta.json()[0]['status'] == 'failure':
|
|
zuul_url = res_sta.json()[0]['target_url']
|
|
o = FailedPR(
|
|
host='gitea',
|
|
url=pull['url'],
|
|
org=gitea_org,
|
|
repo=repo['name'],
|
|
pullrequest=pull['title'],
|
|
status=res_sta.json()[0]['status'],
|
|
zuul_url=zuul_url,
|
|
created_at=pull['created_at'],
|
|
updated_at=res_sta.json()[0]['updated_at'],
|
|
error=1000
|
|
)
|
|
o = add_builds_to_obj(obj=o, url=zuul_url, tenant='gl')
|
|
failed_commits.append(o)
|
|
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res_sta.status_code) +
|
|
" | " + str(res_sta.reason))
|
|
return failed_commits
|
|
|
|
def get_github_repos(url, headers, gh_org):
|
|
"""
|
|
Get all repositories of one GitHub organization
|
|
"""
|
|
repositories = []
|
|
i = 1
|
|
|
|
while True:
|
|
try:
|
|
req_url = url + 'orgs/' + gh_org + '/repos?page=' + str(i)
|
|
res = requests.request('GET', url=req_url, headers=headers)
|
|
if res.json():
|
|
for repo in res.json():
|
|
if repo['archived'] == False:
|
|
repositories.append(repo)
|
|
i+=1
|
|
continue
|
|
else:
|
|
break
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res.status_code) +
|
|
" | " + str(res.reason))
|
|
break
|
|
return repositories
|
|
|
|
def get_github_prs(url, headers, gh_org, repo):
|
|
"""
|
|
Get all Pull Requests of one GitHub repository
|
|
"""
|
|
pullrequests = []
|
|
i = 1
|
|
|
|
while True:
|
|
try:
|
|
req_url = (url + 'repos/' + gh_org + '/' + repo +
|
|
'/pulls?state=open&page=' + str(i))
|
|
res = requests.request('GET', url=req_url, headers=headers)
|
|
if res.json():
|
|
for pr in res.json():
|
|
pullrequests.append(pr)
|
|
i+=1
|
|
continue
|
|
else:
|
|
break
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res.status_code) +
|
|
" | " + str(res.reason))
|
|
break
|
|
return pullrequests
|
|
|
|
|
|
def get_failed_gh_commits(pull, url, gh_org, repo, headers):
|
|
"""
|
|
Collect all Failed Pull Requests of one GitHub repository
|
|
"""
|
|
failed_commits = []
|
|
try:
|
|
req_url = (url + 'repos/' + gh_org + '/' + repo['name'] +
|
|
'/commits/' + pull['head']['sha'] + '/check-runs')
|
|
res_sta = requests.request('GET', url=req_url, headers=headers)
|
|
if res_sta.json():
|
|
if len(res_sta.json()['check_runs']) != 0:
|
|
if res_sta.json()['check_runs'][0]['conclusion'] == 'failure':
|
|
zuul_url = res_sta.json()['check_runs'][0]['details_url']
|
|
o = FailedPR(
|
|
host='github',
|
|
url=pull['html_url'],
|
|
org=gh_org,
|
|
repo=repo['name'],
|
|
pullrequest=pull['title'],
|
|
status=res_sta.json()['check_runs'][0]['conclusion'],
|
|
zuul_url=zuul_url,
|
|
created_at=pull['created_at'],
|
|
updated_at=(res_sta.json()['check_runs']
|
|
[0]['completed_at']),
|
|
error=1000
|
|
)
|
|
o = add_builds_to_obj(obj=o, url=zuul_url, tenant='eco')
|
|
failed_commits.append(o)
|
|
else:
|
|
o = FailedPR(
|
|
host='github',
|
|
url=pull['html_url'],
|
|
org=gh_org,
|
|
repo=repo['name'],
|
|
pullrequest=pull['title'],
|
|
created_at=pull['created_at'],
|
|
updated_at=pull['updated_at'],
|
|
error=1001,
|
|
)
|
|
failed_commits.append(o)
|
|
|
|
except Exception as e:
|
|
print("An error has occured: " + str(e))
|
|
print("The request status is: " + str(res_sta.status_code) +
|
|
" | " + str(res_sta.reason))
|
|
return failed_commits
|
|
|
|
|
|
def create_json_result(failed_commits):
|
|
"""
|
|
Create Result
|
|
"""
|
|
result = {}
|
|
result['meta'] = {}
|
|
result['data'] = []
|
|
if len(failed_commits) != 0:
|
|
failed_commits_json = []
|
|
for o in failed_commits:
|
|
failed_commits_json.append(vars(o))
|
|
result['meta']['count'] = len(failed_commits_json)
|
|
result['data'] = failed_commits_json
|
|
else:
|
|
result['meta']['count'] = 0
|
|
|
|
return json.dumps(result)
|
|
|
|
|
|
def main():
|
|
args = get_args()
|
|
failed_commits = []
|
|
|
|
if args.debug:
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
for h in args.hoster:
|
|
if h == 'gitea':
|
|
|
|
headers = {}
|
|
headers['accept'] = 'application/json'
|
|
if not args.gitea_token:
|
|
raise Exception('Please, provide Gitea Token as argument.')
|
|
if not args.gitea_url:
|
|
raise Exception('Please, provide Gitea URL as argument.')
|
|
headers['Authorization'] = 'token ' + args.gitea_token
|
|
|
|
for org in args.gitea_orgs:
|
|
repos = get_gitea_repos(
|
|
url=args.gitea_url,
|
|
headers=headers,
|
|
gitea_org=org
|
|
)
|
|
for repo in repos:
|
|
pulls = get_gitea_prs(
|
|
url=args.gitea_url,
|
|
headers=headers,
|
|
gitea_org=org,
|
|
repo=repo['name']
|
|
)
|
|
if pulls:
|
|
for pull in pulls:
|
|
commits = get_failed_gitea_commits(
|
|
pull=pull,
|
|
url=args.gitea_url,
|
|
gitea_org=org,
|
|
repo=repo,
|
|
headers=headers
|
|
)
|
|
failed_commits.extend(commits)
|
|
|
|
elif h == 'github':
|
|
if not args.gh_url:
|
|
raise ValueError('Parameter --gh-url not found.')
|
|
url = args.gh_url
|
|
headers = {}
|
|
headers['accept'] = 'application/json'
|
|
if not args.gh_token:
|
|
raise Exception('Please, provide GitHub Token as argument.')
|
|
headers['Authorization'] = 'Bearer ' + args.gh_token
|
|
|
|
for org in args.gh_orgs:
|
|
repos = get_github_repos(
|
|
url=url,
|
|
headers=headers,
|
|
gh_org=org
|
|
)
|
|
for repo in repos:
|
|
pulls = get_github_prs(
|
|
url=url,
|
|
headers=headers,
|
|
gh_org=org,
|
|
repo=repo['name']
|
|
)
|
|
if pulls:
|
|
for pull in pulls:
|
|
commits = get_failed_gh_commits(
|
|
pull=pull,
|
|
url=url,
|
|
gh_org=org,
|
|
repo=repo,
|
|
headers=headers
|
|
)
|
|
failed_commits.extend(commits)
|
|
|
|
print(create_json_result(
|
|
failed_commits=failed_commits))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|