Source Code for Package ndg.saml.test.binding.soap

  1  """NDG SAML SOAP Binding unit test package 
  2   
  3  NERC DataGrid Project 
  4  """ 
  5  __author__ = "P J Kershaw" 
  6  __date__ = "21/08/09" 
  7  __copyright__ = "(C) 2009 Science and Technology Facilities Council" 
  8  __license__ = "http://www.apache.org/licenses/LICENSE-2.0" 
  9  __contact__ = "Philip.Kershaw@stfc.ac.uk" 
 10  __revision__ = '$Id: __init__.py 7154 2010-07-01 15:59:59Z pjkersha $' 
 11  import os 
 12  import unittest 
 13  import socket 
 14  import paste.fixture 
 15  from paste.deploy import loadapp 
 16   
 17  from ndg.soap.test import PasteDeployAppServer 
 18   
 19   
 20 -class TestApp(object): 
 21      """Dummy application to terminate middleware stack containing SAML service 
 22      """ 
 23 -    def __init__(self, global_conf, **app_conf): 
 24          pass 
 25       
 26 -    def __call__(self, environ, start_response): 
 27          response = "404 Not Found" 
 28          start_response(response, 
 29                         [('Content-length', str(len(response))), 
 30                          ('Content-type', 'text/plain')]) 
 31                               
 32          return [response] 
 33   
 34   
 35 -class WithPasteFixtureBaseTestCase(unittest.TestCase): 
 36      """Base class for testing SAML SOAP Binding Query/Response interface 
 37      using a Paste Deploy ini file and Paste Fixture 
 38      """ 
 39      HERE_DIR = os.path.dirname(os.path.abspath(__file__)) 
 40      CONFIG_FILENAME = None # Set in derived class 
 41       
 42 -    def __init__(self, *args, **kwargs): 
 43          wsgiapp = loadapp('config:%s' % self.__class__.CONFIG_FILENAME,  
 44                            relative_to=self.__class__.HERE_DIR) 
 45           
 46          self.app = paste.fixture.TestApp(wsgiapp) 
 47            
 48          unittest.TestCase.__init__(self, *args, **kwargs) 
 49           
 50       
 51 -class WithPasterBaseTestCase(unittest.TestCase): 
 52      """Base class for testing SAML SOAP Binding Query/Response interface 
 53      using a Paste Deploy ini file and Paste Fixture 
 54      """ 
 55      THIS_DIR = os.path.dirname(os.path.abspath(__file__)) 
 56      CONFIG_FILENAME = None # Set in derived class 
 57      SERVICE_PORTNUM = 5443 
 58      SERVER_CERT_FILEPATH = os.path.join(THIS_DIR, 'localhost.crt') 
 59      SERVER_PRIKEY_FILEPATH = os.path.join(THIS_DIR, 'localhost.key') 
 60       
 61 -    def __init__(self, *arg, **kw): 
 62          withSSL = kw.pop('withSSL', False) 
 63          unittest.TestCase.__init__(self, *arg, **kw) 
 64           
 65          self.services = [] 
 66          self.disableServiceStartup = False 
 67          cfgFilePath = os.path.join(self.__class__.THIS_DIR,  
 68                                     self.__class__.CONFIG_FILENAME) 
 69           
 70          self.addService(cfgFilePath=cfgFilePath, 
 71                          withSSL=withSSL, 
 72                          port=self.__class__.SERVICE_PORTNUM) 
 73           
 74 -    def addService(self, *arg, **kw): 
 75          """Utility for setting up threads to run Paste HTTP based services with 
 76          unit tests 
 77           
 78          @param arg: tuple contains ini file path setting for the service 
 79          @type arg: tuple 
 80          @param kw: keywords including "port" - port number to run the service  
 81          from 
 82          @type kw: dict 
 83          """ 
 84          if self.disableServiceStartup: 
 85              return 
 86           
 87          withSSL = kw.pop('withSSL', False) 
 88          if withSSL: 
 89              from OpenSSL import SSL 
 90               
 91              certFilePath = self.__class__.SERVER_CERT_FILEPATH 
 92              priKeyFilePath = self.__class__.SERVER_PRIKEY_FILEPATH 
 93               
 94              kw['ssl_context'] = SSL.Context(SSL.SSLv23_METHOD) 
 95              kw['ssl_context'].set_options(SSL.OP_NO_SSLv2) 
 96           
 97              kw['ssl_context'].use_privatekey_file(priKeyFilePath) 
 98              kw['ssl_context'].use_certificate_file(certFilePath) 
 99           
100          try: 
101              self.services.append(PasteDeployAppServer(*arg, **kw)) 
102              self.services[-1].startThread() 
103               
104          except socket.error: 
105              pass 
106   
107 -    def __del__(self): 
108          """Stop any services started with the addService method and clean up 
109          the CA directory following the trust roots call 
110          """ 
111          if hasattr(self, 'services'): 
112              for service in self.services: 
113                  service.terminateThread() 
114