Source Code for Package ndg.soap.test

 1  """SOAP module unit tests 
 2   
 3  NERC DataGrid Project 
 4  """ 
 5  __author__ = "P J Kershaw" 
 6  __date__ = "24/07/09" 
 7  __copyright__ = "(C) 2009 Science and Technology Facilities Council" 
 8  __contact__ = "Philip.Kershaw@stfc.ac.uk" 
 9  __license__ = "http://www.apache.org/licenses/LICENSE-2.0" 
10  __contact__ = "Philip.Kershaw@stfc.ac.uk" 
11  __revision__ = "$Id: __init__.py 7130 2010-06-30 13:33:07Z pjkersha $" 
12  import logging 
13  logging.basicConfig(level=logging.DEBUG) 
14   
15  import paste.httpserver 
16  from threading import Thread 
17  from paste.deploy import loadapp 
18  from paste.script.util.logging_config import fileConfig 
19   
20   
21 -class PasteDeployAppServer(object): 
22      """Wrapper to paste.httpserver to enable background threading""" 
23       
24 -    def __init__(self, app=None, cfgFilePath=None, port=7443, host='0.0.0.0', 
25                   ssl_context=None, withLoggingConfig=True): 
26          """Load an application configuration from cfgFilePath ini file and  
27          instantiate Paste server object 
28          """        
29          self.__thread = None 
30           
31          if cfgFilePath: 
32              if withLoggingConfig: 
33                  fileConfig(cfgFilePath) 
34              app = loadapp('config:%s' % cfgFilePath) 
35               
36          elif app is None: 
37              raise KeyError('Either the "cfgFilePath" or "app" keyword must be ' 
38                             'set') 
39                          
40          self.__pasteServer = paste.httpserver.serve(app, host=host, port=port,  
41                                                      start_loop=False,  
42                                                      ssl_context=ssl_context) 
43       
44      @property 
45 -    def pasteServer(self): 
46          return self.__pasteServer 
47       
48      @property 
49 -    def thread(self): 
50          return self.__thread 
51       
52 -    def start(self): 
53          """Start server""" 
54          self.pasteServer.serve_forever() 
55           
56 -    def startThread(self): 
57          """Start server in a separate thread""" 
58          self.__thread = Thread(target=PasteDeployAppServer.start, args=(self,)) 
59          self.thread.start() 
60           
61 -    def terminateThread(self): 
62          self.pasteServer.server_close() 
63