Mercurial > hg > orthanc-tests

comparison Tests/Toolbox.py @ 4:292a46fe374c

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
reorganization
author Sebastien Jodogne <s.jodogne@gmail.com>
date Wed, 17 Jun 2015 10:05:27 +0200
parents 2dbba2e6aa4b
children 6d645b3011e1
comparison
equal deleted inserted replaced
3:2dbba2e6aa4b 4:292a46fe374c
19 19
20 20
21 import hashlib 21 import hashlib
22 import httplib2 22 import httplib2
23 import json 23 import json
24 import os.path 24 import os
25 import re 25 import re
26 import signal
26 import subprocess 27 import subprocess
28 import threading
27 import time 29 import time
28 import zipfile 30 import zipfile
29 31
30 from PIL import Image 32 from PIL import Image
31 from urllib import urlencode 33 from urllib import urlencode
188 m = re.search(r'default via ([0-9.]+)', route) 190 m = re.search(r'default via ([0-9.]+)', route)
189 if m == None: 191 if m == None:
190 return 'localhost' 192 return 'localhost'
191 else: 193 else:
192 return m.groups()[0] 194 return m.groups()[0]
195
196
197
198
199 class ExternalCommandThread:
200 @staticmethod
201 def ExternalCommandFunction(arg, stop_event, command, env):
202 external = subprocess.Popen(command, env = env)
203
204 while (not stop_event.is_set()):
205 error = external.poll()
206 if error != None:
207 # http://stackoverflow.com/a/1489838/881731
208 os._exit(-1)
209 stop_event.wait(0.1)
210
211 print 'Stopping the external command'
212 external.terminate()
213
214 def __init__(self, command, env = None):
215 self.thread_stop = threading.Event()
216 self.thread = threading.Thread(target = self.ExternalCommandFunction,
217 args = (10, self.thread_stop, command, env))
218 self.daemon = True
219 self.thread.start()
220
221 def stop(self):
222 self.thread_stop.set()
223 self.thread.join()