changeset 4:292a46fe374c

reorganization
author Sebastien Jodogne <s.jodogne@gmail.com>
date Wed, 17 Jun 2015 10:05:27 +0200
parents 2dbba2e6aa4b
children 19eb87a45685
files Tests/ExternalCommandThread.py Tests/Run.py Tests/Toolbox.py
diffstat 3 files changed, 32 insertions(+), 52 deletions(-) [+]
line wrap: on
line diff
--- a/Tests/ExternalCommandThread.py	Wed Jun 17 10:03:49 2015 +0200
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,50 +0,0 @@
-#!/usr/bin/python
-
-# Orthanc - A Lightweight, RESTful DICOM Store
-# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics
-# Department, University Hospital of Liege, Belgium
-#
-# This program is free software: you can redistribute it and/or
-# modify it under the terms of the GNU General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-# 
-# This program is distributed in the hope that it will be useful, but
-# WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program. If not, see <http://www.gnu.org/licenses/>.
-
-
-import os
-import signal
-import subprocess
-import threading
-
-class ExternalCommandThread:
-    @staticmethod
-    def ExternalCommandFunction(arg, stop_event, command, env):
-        external = subprocess.Popen(command, env = env)
-
-        while (not stop_event.is_set()):
-            error = external.poll()
-            if error != None:
-                # http://stackoverflow.com/a/1489838/881731
-                os._exit(-1)
-            stop_event.wait(0.1)
-
-        print 'Stopping the external command'
-        external.terminate()
-
-    def __init__(self, command, env = None):
-        self.thread_stop = threading.Event()
-        self.thread = threading.Thread(target = self.ExternalCommandFunction, 
-                                       args = (10, self.thread_stop, command, env))
-        self.daemon = True
-        self.thread.start()
-
-    def stop(self):
-        self.thread_stop.set()
-        self.thread.join()
--- a/Tests/Run.py	Wed Jun 17 10:03:49 2015 +0200
+++ b/Tests/Run.py	Wed Jun 17 10:05:27 2015 +0200
@@ -25,7 +25,6 @@
 import unittest
 import pprint
 
-from ExternalCommandThread import *
 from Toolbox import *
 from Tests import *
 
--- a/Tests/Toolbox.py	Wed Jun 17 10:03:49 2015 +0200
+++ b/Tests/Toolbox.py	Wed Jun 17 10:05:27 2015 +0200
@@ -21,9 +21,11 @@
 import hashlib
 import httplib2
 import json
-import os.path
+import os
 import re
+import signal
 import subprocess
+import threading
 import time
 import zipfile
 
@@ -190,3 +192,32 @@
         return 'localhost'
     else:
         return m.groups()[0]
+
+
+
+
+class ExternalCommandThread:
+    @staticmethod
+    def ExternalCommandFunction(arg, stop_event, command, env):
+        external = subprocess.Popen(command, env = env)
+
+        while (not stop_event.is_set()):
+            error = external.poll()
+            if error != None:
+                # http://stackoverflow.com/a/1489838/881731
+                os._exit(-1)
+            stop_event.wait(0.1)
+
+        print 'Stopping the external command'
+        external.terminate()
+
+    def __init__(self, command, env = None):
+        self.thread_stop = threading.Event()
+        self.thread = threading.Thread(target = self.ExternalCommandFunction, 
+                                       args = (10, self.thread_stop, command, env))
+        self.daemon = True
+        self.thread.start()
+
+    def stop(self):
+        self.thread_stop.set()
+        self.thread.join()