-
Notifications
You must be signed in to change notification settings - Fork 0
/
script.py
72 lines (69 loc) · 2.51 KB
/
script.py
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
#!/usr/bin/python3
import os
import sys
import time
import requests
import json
import logging
logging.basicConfig(level=logging.INFO)
token = os.environ['INPUT_TOKEN']
user = os.environ['INPUT_USER']
repo = os.environ['INPUT_REPO']
workflow = os.environ['INPUT_WORKFLOW']
env = os.environ['INPUT_ENV']
group = os.environ['INPUT_GROUP']
branch = os.environ['INPUT_BRANCH']
wait_max_attempt = 60
url_trigger = "https://api.github.com/repos/{0}/{1}/actions/workflows/{2}.yaml/dispatches".format(user, repo, workflow)
payload = json.dumps({
"ref": "{0}".format(branch),
"inputs": {
"target-env": "{0}".format(env),
"test-group": "{0}".format(group),
}
})
headers = {
'Accept': 'application/vnd.github.everest-preview+json',
'Content-Type': 'application/json',
'Authorization': 'Bearer {0}'.format(token)
}
url_dispatch = "https://api.github.com/repos/{0}/{1}/actions/runs".format(user, repo)
headersGet = {
'Accept': 'application/vnd.github.everest-preview+json',
'Authorization': 'Bearer {0}'.format(token)
}
logging.info('Dispatching action')
logging.info(payload)
response = requests.request("POST", url_trigger, headers=headers, data=payload)
if response.status_code != 204:
sys.exit(response.text)
time.sleep(3)
workflowId = ''
response = requests.request("GET", url_dispatch, headers=headersGet)
if response.reason == 'OK':
json = json.loads(response.text)
for node in json['workflow_runs']:
if node['status'] != 'completed':
break
workflowId = node['id']
logging.info("Found build with id {0}".format(workflowId))
counter = 0
while counter < wait_max_attempt:
url = "https://api.github.com/repos/{0}/{1}/actions/runs/{2}".format(user, repo, node['id'])
run = requests.request("GET", url, headers=headersGet)
runJson = run.json()
currentStatus = runJson['status']
logging.info("Current run id: {0} has status: {1} ".format(runJson['id'], runJson['status']))
conclusion = runJson['conclusion']
run_number = runJson['run_number']
if currentStatus == "completed":
if conclusion != 'success':
message = "Run number:{0} concluded with status:{1} with run has been finished".format(run_number, conclusion)
assert False, message
else:
logging.info("Run:{0} has been finished with conclusion:{1}".format(run_number, conclusion))
break
time.sleep(15)
counter += 1
else:
AssertionError("Failed to fetch run")