-
Notifications
You must be signed in to change notification settings - Fork 7
/
test_tool.py
208 lines (170 loc) · 6.62 KB
/
test_tool.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
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
#!/usr/bin/env python
'''
Basic QMENTA Tool Image Local Tester
Usage:
$ ./test_sdk_tool.py image_name inputs/ outputs
$ ./test_sdk_tool.py image_name inputs/ outputs --settings settings.json --values values.json --tool package.tool -v /host/path:/container/path
Compatible with Python 2 & 3 with no additional packages.
See https://docs.qmenta.com/sdk/testing.html for more information.
'''
import argparse
import random
import os
import string
import subprocess
import sys
import time
from os import listdir, system
from os.path import abspath, isdir
def random_seq(n):
'''
Generates a random sequence of upper-cased characters of size n
'''
return ''.join(random.choice(string.ascii_uppercase) for x in range(n))
def parse_arguments():
'''
Interface for script
'''
parser = argparse.ArgumentParser(description='Local execution of a tool image test')
# Positional arguments
parser.add_argument('image', help='The name of the tool image (name:tag)')
parser.add_argument('inputs', help='A folder which will be used as input container')
parser.add_argument('outputs', help='A folder which will be used as output container')
# Keyword arguments
parser.add_argument('-v', help='Mount a directory inside the Docker container', action='append')
parser.add_argument('--settings', help='The settings.json file')
parser.add_argument('--values', help='The values for the settings file')
parser.add_argument('--tool', help='The python file with the run method')
# Private testing arguments
parser.add_argument('--resources', help='argparse.SUPPRESS')
return parser.parse_args()
def run_command(cmd, verbose=True):
'''
Execute command in shell
'''
if verbose:
print(' '.join(cmd))
subprocess.call(cmd)
def error(msg):
'''
Exit with error
'''
print(msg)
sys.exit(1)
def main():
'''
Script entrypoint
'''
args = parse_arguments()
# Check input and output folders
if not isdir(abspath(args.inputs)):
error('Error: Input folder does not exist')
if not isdir(abspath(args.outputs)):
error('Error: Output folder does not exist')
if not os.listdir(abspath(args.outputs)) == []:
print('Warning: Output folder is not empty (files could be overwritten).')
# Post executions actions
yes = {'yes', 'y'}
no = {'no', 'n'}
choice = ''
while choice not in yes and choice not in no:
print("Do you want to continue? (Y/N)")
choice = raw_input().lower()
if choice in no:
sys.exit(1)
if args.resources and not isdir(abspath(args.resources)):
error('Error: Resources folder does not exist')
if args.settings and not args.values:
error('Error: Entering a custom settings file requires a settings values file')
# In-container paths
c_settings_path = '/root/local_exec_settings.json'
c_values_path = '/root/local_exec_settings_values.json'
c_input_path = '/root/local_exec_input/'
c_output_path = '/root/local_exec_output/'
c_res_path = '/root/local_exec_resources/'
# If no tool_settings file is provided, use a generic one
if not args.settings:
settings_content = '''[
{
"type": "container",
"title": "Example Container",
"id":"input",
"mandatory":1,
"batch":1,
"file_filter": "c_files[1,*]<'', [], '.*'>",
"in_filter":["mri_brain_data"],
"out_filter":[],
"anchor":1
}
]'''
args.settings = './generic_settings.json'
with open(args.settings, 'w') as settings_file:
settings_file.write(settings_content)
# If no tool_settings_values file is provided, generate just the list of input files
if not args.values:
files = listdir(os.path.join(abspath(args.inputs), 'input'))
values_content = '{"input":[\n' + ',\n'.join(
[' {"path": "' + f + '"}' for f in files if not isdir(f)]) + ']\n}'
args.values = './generic_values.json'
with open(args.values, 'w') as values_file:
values_file.write(values_content)
if not args.tool:
args.tool = 'tool'
# Extra directories to be mounted (for instance, live version of source code)
extra_volumes = []
for volume in (args.v or []):
extra_volumes += ['-v', volume]
# Launch the detached container
c_name = 'local_test_' + random_seq(5)
print('\nStarting container {}...'.format(c_name))
run_command([
'docker', 'run', '-dit',
'-v', abspath(args.inputs) + ':' + c_input_path,
'-v', abspath(args.outputs) + ':' + c_output_path] + extra_volumes + [
'--entrypoint=/bin/bash',
'--name=' + c_name,
args.image,
])
# Copy the settings files to the container
run_command(['docker', 'cp', abspath(args.settings), c_name + ':' + c_settings_path], verbose = False)
run_command(['docker', 'cp', abspath(args.values), c_name + ':' + c_values_path], verbose = False)
# Changing to local executor
run_command([
'docker', 'exec', c_name,
'/bin/bash', '-c', r"sed -i.bak 's/\<qmenta.sdk\>/&.local/' /root/entrypoint.sh"
], verbose=False)
# Run the local executor
print('\nRunning {}.py:run()...\n'.format(args.tool))
launch_cmd = [
'docker', 'exec', c_name, '/bin/bash', 'entrypoint.sh',
c_settings_path, c_values_path, c_input_path, c_output_path, '--tool-path', args.tool + ':run'
]
# Additional resources (QMENTA)
if args.resources:
run_command(['docker', 'cp', abspath(args.resources), c_name + ':' + c_res_path], verbose = False)
launch_cmd += ['--res-folder', c_res_path]
run_command(launch_cmd)
# Post executions actions
yes = {'yes', 'y'}
no = {'no', 'n'}
choice = ''
while choice not in yes and choice not in no:
print("Do you want to stop the container? (Y/N)")
choice = raw_input().lower()
if choice in yes:
run_command(['docker', 'stop', c_name])
choice = ''
while choice not in yes and choice not in no:
print("Do you want to delete the container? (Y/N)")
choice = raw_input().lower()
if choice in yes:
run_command(['docker', 'rm', c_name], verbose = True)
else:
choice = ''
while choice not in yes and choice not in no:
print("Do you want to attach to the container? (Y/N)")
choice = raw_input().lower()
if choice in yes:
run_command(['docker', 'attach', c_name], verbose = True)
if __name__ == "__main__":
main()