source: TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/wsgi/__init__.py @ 6040

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/wsgi/__init__.py@6040
Revision 6040, 2.0 KB checked in by pjkersha, 11 years ago (diff)
  • Added capability to thread paste SSL service for unit tests
  • added SSL Client authn functionality to SamlCredentialWallet? using ndg.security.common.utils.m2crypto.SSLContextProxy
Line 
1"""WSGI unit test package
2
3NERC DataGrid Project
4"""
5__author__ = "P J Kershaw"
6__date__ = "23/02/09"
7__copyright__ = "(C) 2009 Science and Technology Facilities Council"
8__license__ = "BSD - see LICENSE file in top-level directory"
9__contact__ = "Philip.Kershaw@stfc.ac.uk"
10__revision__ = '$Id$'
11
12import paste.httpserver
13from threading import Thread
14from paste.deploy import loadapp
15from paste.script.util.logging_config import fileConfig
16
17import logging
18logging.basicConfig(level=logging.DEBUG)
19
20class PasteDeployAppServer(object):
21    """Wrapper to paste.httpserver to enable background threading"""
22   
23    def __init__(self, app=None, cfgFilePath=None, port=7443, host='0.0.0.0',
24                 ssl_context=None):
25        """Load an application configuration from cfgFilePath ini file and
26        instantiate Paste server object
27        """       
28        self.__thread = None
29       
30        if cfgFilePath:
31            fileConfig(cfgFilePath)
32            app = loadapp('config:%s' % cfgFilePath)
33           
34        elif app is None:
35            raise KeyError('Either the "cfgFilePath" or "app" keyword must be '
36                           'set')
37                       
38        self.__pasteServer = paste.httpserver.serve(app, host=host, port=port, 
39                                                    start_loop=False, 
40                                                    ssl_context=ssl_context)
41   
42    def _getPasteServer(self):
43        return self.__pasteServer
44   
45    pasteServer = property(fget=_getPasteServer)
46   
47    def _getThread(self):
48        return self.__thread
49   
50    thread = property(fget=_getThread)
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
Note: See TracBrowser for help on using the repository browser.