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
|
#!/usr/bin/env nix
#!nix shell nixpkgs#python3 --command python
from argparse import ArgumentParser
import pathlib
from urllib.request import build_opener, HTTPCookieProcessor, Request
from http.cookiejar import CookieJar
import json
import time
parser = ArgumentParser()
parser.add_argument("-l", "--language", dest="language", type=str,
help="Name of the scripting language", metavar="LANG", required=True)
parser.add_argument("-s", "--script", dest="content_file", type=pathlib.Path,
help="read script contents from FILE", metavar="FILE", required=True)
parser.add_argument("-i", "--input", dest="input_file", type=pathlib.Path,
help="read script arguments from FILE", metavar="FILE", required=True)
args = parser.parse_args()
cookiejar = CookieJar()
cookieprocessor = HTTPCookieProcessor(cookiejar)
http_client = build_opener(cookieprocessor)
admin_token = None
id_admin_workspace = "admins"
login_form = {"email": "admin@windmill.dev", "password": "changeme"}
login_req = Request(
"http://localhost:8001/api/auth/login",
method="POST",
headers={'Content-Type': 'application/json'},
data=json.dumps(login_form).encode('utf-8')
)
with http_client.open(login_req) as response:
assert 200 == response.status, f"Failure {response.status}: Superuser login"
assert any(cookie.name == "token" for cookie in cookiejar)
admin_token = next(cookie.value for cookie in cookiejar if cookie.name == "token")
assert admin_token, "Failed to receive a session key from admin login"
if "python" in args.language:
# Windmill package recipe requires a manually set default python version in the global instance settings
python_version_form = {
# NOTE; Update hardcoded python version below to match the windmill package
"value": "3.12"
}
python_version_req = Request(
"http://localhost:8001/api/settings/global/instance_python_version",
method="GET",
headers={'Authorization': f'Bearer {admin_token}'},
data=json.dumps(python_version_form).encode('utf-8')
)
with http_client.open(python_version_req) as response:
assert 200 == response.status, f"Failure {response.status}: Update global instance python version."
workspace_req = Request(
"http://localhost:8001/api/workspaces/list_as_superadmin",
method="GET",
headers={'Authorization': f'Bearer {admin_token}'}
)
with http_client.open(workspace_req) as response:
assert 200 == response.status, f"Failure {response.status}: List workspaces"
workspace_list = json.loads(response.read().decode('utf-8'))
assert any(workspace['id'] == id_admin_workspace for workspace in workspace_list)
script_hash = None
script_form = {
"path": f"u/admin/{args.language}_test",
"summary": f"Test {args.language}",
"description": "",
"language": args.language,
"content": args.content_file.read_text()
}
script_request = Request(
f"http://localhost:8001/api/w/{id_admin_workspace}/scripts/create",
method="POST",
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {admin_token}',
},
data=json.dumps(script_form).encode('utf-8')
)
with http_client.open(script_request) as response:
assert 201 == response.status, f"Failure {response.status}: Create {args.language} script"
script_hash = response.read().decode('utf-8')
assert script_hash, "Failed to receive an identifier from script creation."
# NOTE; Some languages require dependencies and the depenceny collection tasks take some time to complete
if "bash" not in args.language:
for i in [1, 2, 3, 4, 5, 6]:
time.sleep(10) # seconds
script_request = Request(
f"http://localhost:8001/api/w/{id_admin_workspace}/scripts/deployment_status/h/{script_hash}",
method="GET",
)
with http_client.open(script_request) as response:
try:
assert 200 == response.status, f"Failure {response.status}: Retrieve {args.language} deployment status"
script_metadata = json.loads(response.read().decode('utf-8'))
#
exists_lock_error = bool(script_metadata["lock_error_logs"])
assert not exists_lock_error, "Script deployment did not succeed"
is_deployment_success = script_metadata["lock"] is not None
if is_deployment_success:
break
except AssertionError:
# Re-raise error on final attempt
if i == 6:
raise
job_id = None
request = Request(
f"http://localhost:8001/api/w/{id_admin_workspace}/jobs/run/h/{script_hash}",
method="POST",
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {admin_token}',
},
data=args.input_file.read_bytes(),
)
with http_client.open(request) as response:
assert 201 == response.status, f"Failure {response.status}: Run {args.language} script"
job_id = response.read().decode('utf-8')
assert job_id, "Failed to receive an identifier from job creation/scheduling."
started_jobs = set([job_id])
# NOTE; Some languages require script compilation and take longer to run until completion
timeout = 60 # seconds
timeout_end = time.time() + timeout
while any(started_jobs) and time.time() < timeout_end:
time.sleep(10) # seconds
retrieve_jobs_req = Request(
f"http://localhost:8001/api/w/{id_admin_workspace}/jobs/completed/list",
method="GET",
headers={
'Authorization': f'Bearer {admin_token}',
},
)
with http_client.open(retrieve_jobs_req) as response:
assert 200 == response.status, f"Failure {response.status}: Retrieve jobs"
job_results = json.loads(response.read().decode('utf-8'))
for id in set(started_jobs): # Must create copy of set being iterated over
if any(job['id'] == id for job in job_results if bool(job['success'])):
started_jobs.remove(id)
if any(started_jobs):
# There are started jobs that have not completed before timeout
exit(1)
|
