зеркало из https://github.com/microsoft/LightGBM.git
98 строки
2.6 KiB
Python
98 строки
2.6 KiB
Python
# coding: utf-8
|
|
"""Get the most recent status of workflow for the current PR.
|
|
|
|
[usage]
|
|
python get-workflow-status.py TRIGGER_PHRASE
|
|
|
|
TRIGGER_PHRASE: Code phrase that triggers workflow.
|
|
"""
|
|
|
|
import json
|
|
from os import environ
|
|
from sys import argv, exit
|
|
from time import sleep
|
|
from urllib import request
|
|
|
|
|
|
def get_runs(trigger_phrase):
|
|
"""Get all triggering workflow comments in the current PR.
|
|
|
|
Parameters
|
|
----------
|
|
trigger_phrase : str
|
|
Code phrase that triggers workflow.
|
|
|
|
Returns
|
|
-------
|
|
pr_runs : list
|
|
List of comment objects sorted by the time of creation in decreasing order.
|
|
"""
|
|
pr_runs = []
|
|
if environ.get("GITHUB_EVENT_NAME", "") == "pull_request":
|
|
pr_number = int(environ.get("GITHUB_REF").split("/")[-2])
|
|
page = 1
|
|
while True:
|
|
req = request.Request(
|
|
url="{}/repos/microsoft/LightGBM/issues/{}/comments?page={}&per_page=100".format(
|
|
environ.get("GITHUB_API_URL"), pr_number, page
|
|
),
|
|
headers={"Accept": "application/vnd.github.v3+json"},
|
|
)
|
|
url = request.urlopen(req)
|
|
data = json.loads(url.read().decode("utf-8"))
|
|
url.close()
|
|
if not data:
|
|
break
|
|
runs_on_page = [
|
|
i
|
|
for i in data
|
|
if i["author_association"].lower() in {"owner", "member", "collaborator"}
|
|
and i["body"].startswith("/gha run {}".format(trigger_phrase))
|
|
]
|
|
pr_runs.extend(runs_on_page)
|
|
page += 1
|
|
return pr_runs[::-1]
|
|
|
|
|
|
def get_status(runs):
|
|
"""Get the most recent status of workflow for the current PR.
|
|
|
|
Parameters
|
|
----------
|
|
runs : list
|
|
List of comment objects sorted by the time of creation in decreasing order.
|
|
|
|
Returns
|
|
-------
|
|
status : str
|
|
The most recent status of workflow.
|
|
Can be 'success', 'failure' or 'in-progress'.
|
|
"""
|
|
status = "success"
|
|
for run in runs:
|
|
body = run["body"]
|
|
if "Status: " in body:
|
|
if "Status: skipped" in body:
|
|
continue
|
|
if "Status: failure" in body:
|
|
status = "failure"
|
|
break
|
|
if "Status: success" in body:
|
|
status = "success"
|
|
break
|
|
else:
|
|
status = "in-progress"
|
|
break
|
|
return status
|
|
|
|
|
|
if __name__ == "__main__":
|
|
trigger_phrase = argv[1]
|
|
while True:
|
|
status = get_status(get_runs(trigger_phrase))
|
|
if status != "in-progress":
|
|
break
|
|
sleep(60)
|
|
if status == "failure":
|
|
exit(1)
|