comparison Plugins/WebDav/Run.py @ 338:edaeb57bf01f

creating tests for webdav
author Sebastien Jodogne <s.jodogne@gmail.com>
date Tue, 13 Oct 2020 12:00:08 +0200
parents
children 44ab919be2e9
comparison
equal deleted inserted replaced
337:ec13ace43bde 338:edaeb57bf01f
1 #!/usr/bin/python
2
3 # Orthanc - A Lightweight, RESTful DICOM Store
4 # Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics
5 # Department, University Hospital of Liege, Belgium
6 # Copyright (C) 2017-2020 Osimis S.A., Belgium
7 #
8 # This program is free software: you can redistribute it and/or
9 # modify it under the terms of the GNU General Public License as
10 # published by the Free Software Foundation, either version 3 of the
11 # License, or (at your option) any later version.
12 #
13 # This program is distributed in the hope that it will be useful, but
14 # WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 # General Public License for more details.
17 #
18 # You should have received a copy of the GNU General Public License
19 # along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21
22
23 import argparse
24 import easywebdav
25 import io
26 import os
27 import pprint
28 import re
29 import sys
30 import unittest
31
32 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'Tests'))
33 from Toolbox import *
34
35
36 ##
37 ## Parse the command-line arguments
38 ##
39
40 parser = argparse.ArgumentParser(description = 'Run the integration tests for the patient recycling behavior.')
41
42 parser.add_argument('--server',
43 default = 'localhost',
44 help = 'Address of the Orthanc server to test')
45 parser.add_argument('--rest',
46 type = int,
47 default = 8042,
48 help = 'Port to the REST API')
49 parser.add_argument('--username',
50 default = 'alice',
51 help = 'Username to the REST API')
52 parser.add_argument('--password',
53 default = 'orthanctest',
54 help = 'Password to the REST API')
55 parser.add_argument('--force', help = 'Do not warn the user',
56 action = 'store_true')
57 parser.add_argument('options', metavar = 'N', nargs = '*',
58 help='Arguments to Python unittest')
59
60 args = parser.parse_args()
61
62
63 ##
64 ## Configure the testing context
65 ##
66
67 if not args.force:
68 print("""
69 WARNING: This test will remove all the content of your
70 Orthanc instance running on %s!
71
72 Are you sure ["yes" to go on]?""" % args.server)
73
74 if sys.stdin.readline().strip() != 'yes':
75 print('Aborting...')
76 exit(0)
77
78
79 ORTHANC = DefineOrthanc(server = args.server,
80 username = args.username,
81 password = args.password,
82 restPort = args.rest)
83
84 WEBDAV = easywebdav.connect(args.server,
85 port = args.rest,
86 username = args.username,
87 password = args.password)
88
89
90 ##
91 ## The tests
92 ##
93
94
95 def ListFiles(path, recursive):
96 result = [ ]
97 for i in WEBDAV.ls(path):
98 if i.name == path:
99 pass
100 elif i.contenttype == '':
101 if recursive:
102 result += ListFiles(i.name + '/', True)
103 else:
104 result.append(i.name)
105 return result
106
107
108 def GetFileInfo(path):
109 for i in WEBDAV.ls(path[0 : path.rfind('/')]):
110 if i.name == path:
111 return i
112 raise Exception('Cannot find: %s' % path)
113
114
115 def DownloadFile(path):
116 with tempfile.NamedTemporaryFile(delete = False) as f:
117 f.close()
118 WEBDAV.download(path, f.name)
119 with open(f.name, 'rb') as g:
120 result = g.read()
121 os.unlink(f.name)
122 return result
123
124
125 class Orthanc(unittest.TestCase):
126 def setUp(self):
127 if (sys.version_info >= (3, 0)):
128 # Remove annoying warnings about unclosed socket in Python 3
129 import warnings
130 warnings.simplefilter("ignore", ResourceWarning)
131
132 DropOrthanc(ORTHANC)
133
134
135 def test_root(self):
136 self.assertEqual(6, len(WEBDAV.ls('/webdav/')))
137 for i in WEBDAV.ls('/webdav/'):
138 self.assertTrue(i.name in [
139 '/webdav/',
140 '/webdav/by-dates',
141 '/webdav/by-patients',
142 '/webdav/by-studies',
143 '/webdav/by-uids',
144 '/webdav/uploads'
145 ])
146 self.assertEqual(0, i.size)
147 self.assertEqual('', i.contenttype)
148
149 patients = WEBDAV.ls('/webdav/by-patients/')
150 self.assertEqual(1, len(patients))
151 self.assertEqual(patients[0].name, '/webdav/by-patients/')
152 self.assertEqual(0, patients[0].size)
153 self.assertEqual('', patients[0].contenttype)
154
155
156 def test_upload(self):
157 self.assertEqual(0, len(ListFiles('/webdav/uploads/', True)))
158
159 uploads = WEBDAV.ls('/webdav/uploads/')
160 for i in uploads:
161 self.assertEqual(i.contenttype, '') # Only folders are allowed
162
163 self.assertEqual(0, len(DoGet(ORTHANC, '/studies')))
164 WEBDAV.upload(GetDatabasePath('DummyCT.dcm'), '/webdav/uploads/DummyCT.dcm')
165
166 while len(ListFiles('/webdav/uploads/', True)) > 1:
167 time.sleep(0.1)
168
169 instances = DoGet(ORTHANC, '/instances?expand')
170 self.assertEqual(1, len(instances))
171 self.assertEqual('1.2.840.113619.2.176.2025.1499492.7040.1171286242.109',
172 instances[0]['MainDicomTags']['SOPInstanceUID'])
173
174
175 def test_upload_folders(self):
176 self.assertEqual(0, len(ListFiles('/webdav/uploads/', True)))
177 self.assertEqual(0, len(DoGet(ORTHANC, '/studies')))
178
179 try:
180 WEBDAV.mkdir('/webdav/uploads/a')
181 except:
182 pass
183 try:
184 WEBDAV.mkdir('/webdav/uploads/b')
185 except:
186 pass
187
188 WEBDAV.upload(GetDatabasePath('DummyCT.dcm'), '/webdav/uploads/a/DummyCT.dcm')
189 WEBDAV.upload(GetDatabasePath('ColorTestMalaterre.dcm'), '/webdav/uploads/b/ColorTestMalaterre.dcm')
190
191 while len(ListFiles('/webdav/uploads/', True)) > 1:
192 time.sleep(0.1)
193
194 self.assertEqual(2, len(DoGet(ORTHANC, '/instances')))
195
196
197 def test_by_uids(self):
198 self.assertEqual(1, len(WEBDAV.ls('/webdav/by-uids/')))
199 self.assertEqual(0, len(ListFiles('/webdav/by-uids/', True)))
200 self.assertEqual(0, len(DoGet(ORTHANC, '/instances')))
201
202 i = UploadInstance(ORTHANC, 'DummyCT.dcm')['ID']
203 instance = DoGet(ORTHANC, '/instances/%s/tags?simplify' % i)
204 studyUid = instance['StudyInstanceUID']
205 seriesUid = instance['SeriesInstanceUID']
206 sopUid = instance['SOPInstanceUID']
207
208 self.assertEqual(0, len(ListFiles('/webdav/by-uids/', False)))
209 self.assertEqual(1, len(ListFiles('/webdav/by-uids/%s/' % studyUid, False)))
210 self.assertEqual(2, len(ListFiles('/webdav/by-uids/%s/%s/' % (studyUid, seriesUid), False)))
211
212 content = ListFiles('/webdav/by-uids/', True)
213 self.assertEqual(3, len(content))
214 self.assertTrue(('/webdav/by-uids/%s/study.json' % studyUid) in content)
215 self.assertTrue(('/webdav/by-uids/%s/%s/series.json' % (studyUid, seriesUid)) in content)
216 self.assertTrue(('/webdav/by-uids/%s/%s/%s.dcm' % (studyUid, seriesUid, sopUid)) in content)
217
218 info = GetFileInfo('/webdav/by-uids/%s/study.json' % studyUid)
219 self.assertEqual(info.contenttype, 'application/json')
220
221 info = GetFileInfo('/webdav/by-uids/%s/%s/series.json' % (studyUid, seriesUid))
222 self.assertEqual(info.contenttype, 'application/json')
223
224 info = GetFileInfo('/webdav/by-uids/%s/%s/%s.dcm' % (studyUid, seriesUid, sopUid))
225 self.assertEqual(info.contenttype, 'application/dicom')
226 self.assertEqual(info.size, os.stat(GetDatabasePath('DummyCT.dcm')).st_size)
227
228 a = DownloadFile('/webdav/by-uids/%s/%s/%s.dcm' % (studyUid, seriesUid, sopUid))
229 self.assertEqual(len(a), info.size)
230
231 with open(GetDatabasePath('DummyCT.dcm'), 'rb') as f:
232 self.assertEqual(a, f.read())
233
234 self.assertEqual(studyUid, json.loads(DownloadFile('/webdav/by-uids/%s/study.json' % studyUid))
235 ['MainDicomTags']['StudyInstanceUID'])
236
237 self.assertEqual(seriesUid, json.loads(DownloadFile('/webdav/by-uids/%s/%s/series.json' % (studyUid, seriesUid)))
238 ['MainDicomTags']['SeriesInstanceUID'])
239
240 self.assertEqual(1, len(DoGet(ORTHANC, '/instances')))
241 WEBDAV.delete('/webdav/by-uids/%s/%s/%s.dcm' % (studyUid, seriesUid, sopUid))
242 self.assertEqual(0, len(DoGet(ORTHANC, '/instances')))
243
244 try:
245 print('\nStarting the tests...')
246 unittest.main(argv = [ sys.argv[0] ] + args.options)
247
248 finally:
249 print('\nDone')