-
Notifications
You must be signed in to change notification settings - Fork 2
/
testDataServer.py
269 lines (228 loc) · 12.5 KB
/
testDataServer.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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
# Copyright (c) General Electric Company, 2017. All rights reserved.
# Prototype code for testing dataServer.py
import unittest,json,requests,uuid,tarfile,glob,os,logging
import sys
print(sys.argv)
idx = sys.argv.index('testDataServer.py')
sys.argv = sys.argv[idx:]
print(sys.argv)
from dataServer import app
import shutil
PATIENT = "Adam"
STUDY = "1.2.826.0.1.3680043.2.1125.1.75064541463040.2005072610175534421"
SERIES = "017f45ac-ce32-4921-8700-89d6b0bedb21"
SERIES_PATH = "/Patients/Adam/Primary/Imaging/"+ STUDY + '/' + SERIES
INSTANCE_PATH = "/Patients/Adam/Primary/Imaging/"+ STUDY + '/' + SERIES + '/' + 'vhm.1001.dcm'
UPLOAD_PATH = '/Patients/'+ PATIENT + '/Results/test_pipeline/test_execid/Imaging/test_study'
class TestDataServerAPIs(unittest.TestCase):
def setUp(self): # NOSONAR
self.app = app.test_client()
logging.info("start testing dataserver...")
def tearDown(self): # NOSONAR
pass
def test_not_found(self):
url = 'http://localhost:5106/invalidinput'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 404)
def test_health_check(self):
url = 'http://localhost:5106/v1/health'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
def test_health(self):
url = 'http://localhost:5106/'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
def test_get_patient_list(self):
url = 'http://localhost:5106/v1/patients'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
patients = [p['id'] for p in json_res]
self.assertTrue(len(patients) > 0)
def test_get_patient_info(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
primary_type = json_res['primary']
self.assertEqual(primary_type[0], 'Imaging')
def test_get_study_list(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT + '/imaging/studies'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res[0]['eid'], 'primary')
def test_get_study_type(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT + '/imaging/studies/' + STUDY + '/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res[0], 'series')
def test_get_primary_series_list(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT + '/imaging/studies/' + STUDY + '/primary'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res[0]['eid'], 'primary')
self.assertTrue(len(json_res[0]['id'])>0)
def test_get_series_list(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT + '/imaging/studies/' + STUDY + '/series'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res[0]['eid'], 'primary')
self.assertTrue(len(json_res[0]['id'])>0)
def test_get_series_type(self):
url = 'http://localhost:5106/v1/series' + SERIES_PATH +'/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
def test_get_image_list(self):
url = 'http://localhost:5106/v1/series' + SERIES_PATH +'/instances'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertTrue(len(json_res['paths'])>0)
def test_get_uploading_path(self):
url = 'http://localhost:5106/v1/patients/' + PATIENT + '/results/test_pipeline/steps/test_execid/imaging/studies/test_study/series'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res['path'],UPLOAD_PATH)
def test_retrieve_series(self):
url = 'http://localhost:5106/v1/series'+ SERIES_PATH + '/archive'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertTrue(any(response.data))
def test_upload_series(self):
tar = tarfile.open('./output.tar','w')
for f in glob.glob('/rt106/data' + SERIES_PATH + '/'):
filename = os.path.basename(f)
tar.add(f,arcname=filename)
tar.close()
archive = { 'file' : open('./output.tar' ,'rb') }
upload_path = '/uploads/Radiology/test'
url = 'http://localhost:5106/v1/series' + upload_path + '/tar'
response = self.app.post(url,data=archive)
self.assertEqual(response.status_code, 200)
os.remove('./output.tar')
shutil.rmtree('/rt106/data' + upload_path)
shutil.rmtree('/rt106/data/uploads/')
def test_get_instance_type(self):
url = 'http://localhost:5106/v1/instance' + INSTANCE_PATH +'/type'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual(json_res['type'],'DICOM')
def test_get_instance(self):
url = 'http://localhost:5106/v1/instance' + INSTANCE_PATH +'/DICOM'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertTrue(any(response.data))
def test_upload_instance(self):
url = 'http://localhost:5106/v1/instance/uploads/Pathology/test1/tiff16'
filedir = '/rt106/data/Slides/AGA_260_3/021/Source/DAPI/pERK_CD31_AGA_260_3_S001_P021_dapi.tif'
shutil.copyfile(filedir, './DAPI.tif')
archive = { 'file' : open('./DAPI.tif' ,'rb') }
response = self.app.post(url,data=archive)
self.assertEqual(response.status_code, 200)
self.assertTrue(os.path.isfile('/rt106/data/uploads/Pathology/test1/DAPI.tif'))
shutil.rmtree('/rt106/data/uploads/Pathology/test1/')
shutil.rmtree('/rt106/data/uploads/')
def test_upload_instance_force(self):
url = 'http://localhost:5106/v1/instance/uploads/Pathology/test2/tiff16/force'
filedir = '/rt106/data/Slides/AGA_260_3/021/Source/DAPI/pERK_CD31_AGA_260_3_S001_P021_dapi.tif'
shutil.copyfile(filedir, './DAPI.tif')
archive = { 'file' : open('./DAPI.tif' ,'rb') }
response = self.app.post(url,data=archive)
self.assertEqual(response.status_code, 200)
self.assertTrue(os.path.isfile('/rt106/data/uploads/Pathology/test2/DAPI.tif'))
shutil.rmtree('/rt106/data/uploads/Pathology/test2/')
shutil.rmtree('/rt106/data/uploads/')
def test_get_annotation_type(self):
url = 'http://localhost:5106/v1/annotation' + SERIES_PATH + '/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
def test_get_annotation(self):
url = 'http://localhost:5106/v1/annotation' + SERIES_PATH + '/JSON'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertTrue(any(json_res))
#
# Pathology / microscopy section
#
def test_get_slide_list(self):
url = 'http://localhost:5106/v1/pathology/slides'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn("AGA_260_3", json_res)
def test_get_slide_type(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn('regions', json_res)
def test_get_slide_regions(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn('021', json_res)
def test_get_region_type(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn('channels', json_res)
def test_get_slide_channels(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/channels'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn('DAPI', json_res)
def test_get_channel_type(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/channels/DAPI/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual('tiff16', json_res[0])
def test_get_image_path(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/channels/DAPI/image'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertTrue(len(json_res[0]) > 0)
def test_get_result_types(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/results/test_pipeline/types'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertEqual('steps', json_res[0])
def test_get_result_format(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/results/Source/steps/DAPI'
# need to have the result in the directory to test on
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertIn('tiff16', json_res)
def test_get_result_path(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/results/Source/steps/DAPI/data'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertTrue(len(json_res) > 0)
def test_get_result_image_path(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/results/Source/steps/DAPI/instances'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
json_res = json.loads(response.data)
self.assertTrue(len(json_res) > 0)
def test_get_pipeline_list(self):
url = 'http://localhost:5106/v1/pathology/slides/AGA_260_3/regions/021/results'
response = self.app.get(url,content_type='application/json')
self.assertEqual(response.status_code, 200)
# need to have at least one pipeline available for results
#def test_get_execution_list(self):
if __name__ == '__main__':
unittest.main()