Mercurial > hg > orthanc-tests
comparison Plugins/Worklists/Run.py @ 90:afbac3eb28a5
integration tests of worklists
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Tue, 06 Dec 2016 11:59:35 +0100 |
parents | |
children | ba5619c3941d |
comparison
equal
deleted
inserted
replaced
89:74b7b9aed5f8 | 90:afbac3eb28a5 |
---|---|
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 # | |
7 # This program is free software: you can redistribute it and/or | |
8 # modify it under the terms of the GNU General Public License as | |
9 # published by the Free Software Foundation, either version 3 of the | |
10 # License, or (at your option) any later version. | |
11 # | |
12 # This program is distributed in the hope that it will be useful, but | |
13 # WITHOUT ANY WARRANTY; without even the implied warranty of | |
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU | |
15 # General Public License for more details. | |
16 # | |
17 # You should have received a copy of the GNU General Public License | |
18 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
19 | |
20 | |
21 import argparse | |
22 import os | |
23 import pprint | |
24 import re | |
25 import subprocess | |
26 import sys | |
27 import tempfile | |
28 import unittest | |
29 | |
30 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'Tests')) | |
31 from Toolbox import * | |
32 | |
33 | |
34 ## | |
35 ## Parse the command-line arguments | |
36 ## | |
37 | |
38 parser = argparse.ArgumentParser(description = 'Run the integration tests for the DICOM worklist plugin.') | |
39 | |
40 parser.add_argument('--server', | |
41 default = 'localhost', | |
42 help = 'Address of the Orthanc server to test') | |
43 parser.add_argument('--dicom', | |
44 type = int, | |
45 default = 4242, | |
46 help = 'DICOM port of the Orthanc instance to test') | |
47 parser.add_argument('options', metavar = 'N', nargs = '*', | |
48 help='Arguments to Python unittest') | |
49 | |
50 args = parser.parse_args() | |
51 | |
52 | |
53 | |
54 ## | |
55 ## Toolbox | |
56 ## | |
57 | |
58 DATABASE = os.path.abspath(os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'Database', 'Worklists'))) | |
59 WORKING = os.path.join(DATABASE, 'Working') | |
60 | |
61 try: | |
62 os.mkdir(WORKING) | |
63 except Exception as e: | |
64 # The working folder has already been created | |
65 pass | |
66 | |
67 def ClearDatabase(): | |
68 for f in os.listdir(WORKING): | |
69 if f != 'lockfile': | |
70 os.remove(os.path.join(WORKING, f)) | |
71 | |
72 def AddToDatabase(source): | |
73 subprocess.check_call([ 'dump2dcm', '-g', '--write-xfer-little', | |
74 os.path.join(DATABASE, source), | |
75 os.path.join(WORKING, os.path.basename(source) + '.wl') ]) | |
76 | |
77 def RunQuery(source, ignoreTags): | |
78 with tempfile.NamedTemporaryFile() as f: | |
79 subprocess.check_call([ 'dump2dcm', '-g', '--write-xfer-little', | |
80 os.path.join(DATABASE, source), f.name ]) | |
81 | |
82 a = subprocess.check_output([ 'findscu', '-v', '--call', 'ORTHANC', '-aet', 'ORTHANCTEST', | |
83 args.server, str(args.dicom), f.name ], | |
84 stderr = subprocess.STDOUT).splitlines() | |
85 | |
86 if len(filter(lambda x: x.startswith('E:'), a)) > 0: | |
87 raise Exception('Error while running findscu') | |
88 | |
89 b = map(lambda x: x[3:], filter(lambda x: x.startswith('W: ---') or x.startswith('W: ('), a)) | |
90 b = map(lambda x: re.sub(r'\s*#.*', '', x), b) | |
91 | |
92 answers = [] | |
93 current = [] | |
94 | |
95 for l in b: | |
96 if l[0] == '-': | |
97 # This is a separator between DICOM datasets | |
98 if len(current) > 0: | |
99 answers.append(current) | |
100 current = [] | |
101 | |
102 else: | |
103 tag = l[1:10].lower() | |
104 if not tag in ignoreTags: | |
105 current.append(l) | |
106 | |
107 if len(current) > 0: | |
108 answers.append(current) | |
109 | |
110 return answers | |
111 | |
112 def CompareAnswers(expected, actual): | |
113 if len(expected) != len(actual): | |
114 return False | |
115 | |
116 if len(expected) == 0: | |
117 return True | |
118 | |
119 for i in range(len(expected)): | |
120 for j in range(len(actual)): | |
121 if expected[i] == actual[j]: | |
122 return True | |
123 | |
124 return False | |
125 | |
126 | |
127 ## | |
128 ## The tests | |
129 ## | |
130 | |
131 class Orthanc(unittest.TestCase): | |
132 def setUp(self): | |
133 if (sys.version_info >= (3, 0)): | |
134 # Remove annoying warnings about unclosed socket in Python 3 | |
135 import warnings | |
136 warnings.simplefilter("ignore", ResourceWarning) | |
137 | |
138 ClearDatabase() | |
139 | |
140 | |
141 def test_single(self): | |
142 for db in range(1, 11): | |
143 ClearDatabase() | |
144 AddToDatabase('DcmtkDatabase/wklist%d.dump' % db) | |
145 | |
146 for query in range(0, 13): | |
147 answers = RunQuery('DcmtkQueries/wlistqry%d.dump' % query, [ | |
148 '0008,0005', | |
149 '0040,0004', | |
150 '0040,0005', | |
151 '0040,0020', | |
152 ]) | |
153 | |
154 with open(os.path.join('%s/DcmtkExpected/single-%d-%d.json' % (DATABASE, db, query)), 'r') as f: | |
155 expected = json.loads(f.read()) | |
156 self.assertTrue(CompareAnswers(expected, answers)) | |
157 | |
158 | |
159 def test_all(self): | |
160 ClearDatabase() | |
161 | |
162 for db in range(1, 11): | |
163 AddToDatabase('DcmtkDatabase/wklist%d.dump' % db) | |
164 | |
165 for query in range(0, 13): | |
166 answers = RunQuery('DcmtkQueries/wlistqry%d.dump' % query, [ | |
167 '0008,0005', | |
168 '0040,0004', | |
169 '0040,0005', | |
170 '0040,0020', | |
171 ]) | |
172 | |
173 with open(os.path.join('%s/DcmtkExpected/all-%d.json' % (DATABASE, query)), 'r') as f: | |
174 expected = json.loads(f.read()) | |
175 self.assertTrue(CompareAnswers(expected, answers)) | |
176 | |
177 | |
178 | |
179 try: | |
180 print('\nStarting the tests...') | |
181 unittest.main(argv = [ sys.argv[0] ] + args.options) | |
182 | |
183 finally: | |
184 print('\nDone') |