source: TI12-security/trunk/ndg_saml/ndg/soap/test/test_soap.py @ 7662

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/ndg_saml/ndg/soap/test/test_soap.py@7662
Revision 7662, 7.4 KB checked in by pjkersha, 10 years ago (diff)

Added SOAPFault handling for ndg.soap package. Needs integration into ndg.saml.saml2.binding.soap.server.wsgi.queryinterface to enable SAML query interface to do better error reporting.

Line 
1#!/usr/bin/env python
2"""SOAP module unit test module
3
4NERC DataGrid Project
5"""
6__author__ = "P J Kershaw"
7__date__ = "24/07/09"
8__copyright__ = "(C) 2009 Science and Technology Facilities Council"
9__contact__ = "Philip.Kershaw@stfc.ac.uk"
10__license__ = "http://www.apache.org/licenses/LICENSE-2.0"
11__contact__ = "Philip.Kershaw@stfc.ac.uk"
12__revision__ = "$Id: test_soap.py 7134 2010-06-30 13:49:40Z pjkersha $"
13import logging
14logging.basicConfig(level=logging.DEBUG)
15
16import unittest
17import socket
18from cStringIO import StringIO
19from os import path
20import paste.fixture
21from urllib2 import HTTPHandler, URLError
22
23from ndg.soap import SOAPFaultBase
24from ndg.soap.etree import SOAPEnvelope, SOAPFault, SOAPFaultException
25from ndg.soap.client import UrlLib2SOAPClient, UrlLib2SOAPRequest
26from ndg.soap.test import PasteDeployAppServer
27
28
29class SOAPBindingMiddleware(object):
30    """Simple WSGI interface for SOAP service"""
31       
32    def __call__(self, environ, start_response):
33        requestFile = environ['wsgi.input']
34       
35        print("Server received request from client:\n\n%s" % 
36              requestFile.read())
37       
38        soapResponse = SOAPEnvelope()
39        soapResponse.create()
40       
41        response = soapResponse.serialize()
42        start_response("200 OK",
43                       [('Content-length', str(len(response))),
44                        ('Content-type', 'text/xml')])
45        return [response]
46   
47   
48class SOAPTestCase(unittest.TestCase):
49    EG_SOAPFAULT_CODE = "%s:%s" % (SOAPFaultBase.DEFAULT_ELEMENT_NS_PREFIX, 
50                                   "MustUnderstand")
51    EG_SOAPFAULT_STRING = "Can't process element X set with mustUnderstand"
52       
53    def test01Envelope(self):
54        envelope = SOAPEnvelope()
55        envelope.create()
56        soap = envelope.serialize()
57       
58        self.assert_(len(soap) > 0)
59        self.assert_("Envelope" in soap)
60        self.assert_("Body" in soap)
61        self.assert_("Header" in soap)
62       
63        print(envelope.prettyPrint())
64        stream = StringIO()
65        stream.write(soap)
66        stream.seek(0)
67       
68        envelope2 = SOAPEnvelope()
69        envelope2.parse(stream)
70        soap2 = envelope2.serialize()
71        self.assert_(soap2 == soap)
72
73    def test02CreateSOAPFaultBase(self):
74       
75        fault = SOAPFaultBase(self.__class__.EG_SOAPFAULT_STRING, 
76                              self.__class__.EG_SOAPFAULT_CODE)
77       
78        self.assert_(fault.faultCode == self.__class__.EG_SOAPFAULT_CODE)
79        self.assert_(fault.faultString == self.__class__.EG_SOAPFAULT_STRING)
80     
81    def _createSOAPFault(self):
82        fault = SOAPFault(self.__class__.EG_SOAPFAULT_STRING, 
83                          self.__class__.EG_SOAPFAULT_CODE)
84        fault.create()
85        return fault   
86   
87    def test03SerialiseSOAPFault(self):
88        # Use ElementTree implementation
89        fault = self._createSOAPFault()
90        faultStr = fault.serialize()
91        print(faultStr)
92        self.assert_(self.__class__.EG_SOAPFAULT_STRING in faultStr)
93
94    def test04ParseSOAPFault(self):
95        fault = self._createSOAPFault()
96        faultStr = fault.serialize()
97        stream = StringIO()
98        stream.write(faultStr)
99        stream.seek(0)
100       
101        fault2 = SOAPFault()
102        fault2.parse(stream)
103        self.assert_(fault2.faultCode == fault.faultCode)
104        self.assert_(fault2.faultString == fault.faultString)
105   
106    def test05CreateSOAPFaultException(self):
107        try:
108            raise SOAPFaultException("bad request", SOAPFault.CLIENT_FAULT_CODE)
109       
110        except SOAPFaultException, e:
111            self.assert_(e.fault.faultString == "bad request")
112            self.assert_(SOAPFault.CLIENT_FAULT_CODE in e.fault.faultCode)
113            e.fault.create()
114            self.assert_("bad request" in e.fault.serialize())
115            self.assert_(SOAPFault.CLIENT_FAULT_CODE in e.fault.serialize())
116            return
117       
118        self.fail("Expecting SOAPFaultException raised")
119       
120    def test06CreateSOAPFaultResponse(self):
121        # Create full SOAP Response containing a SOAP Fault
122        envelope = SOAPEnvelope()
123        envelope.body.fault = self._createSOAPFault()
124        envelope.create()
125        soap = envelope.serialize()
126       
127        self.assert_(len(soap) > 0)
128        self.assert_("Envelope" in soap)
129        self.assert_("Body" in soap)
130        self.assert_("Header" in soap)
131        self.assert_("Fault" in soap)
132       
133        print(envelope.prettyPrint())
134        stream = StringIO()
135        stream.write(soap)
136        stream.seek(0)
137       
138        envelope2 = SOAPEnvelope()
139        envelope2.parse(stream)
140        soap2 = envelope2.serialize()
141        self.assert_(soap2 == soap)
142           
143   
144class SOAPServiceTestCase(unittest.TestCase):
145    SOAP_SERVICE_PORTNUM = 10080
146    ENDPOINT = 'http://localhost:%d/soap' % SOAP_SERVICE_PORTNUM
147    THIS_DIR = path.abspath(path.dirname(__file__))   
148   
149    def __init__(self, *args, **kwargs):
150        """Use paste.fixture to test client/server SOAP interface"""
151        self.services = []
152        self.disableServiceStartup = False
153
154        wsgiApp = SOAPBindingMiddleware()
155        self.app = paste.fixture.TestApp(wsgiApp)
156         
157        super(SOAPServiceTestCase, self).__init__(*args, **kwargs)
158             
159    def test01SendRequest(self):
160        requestEnvelope = SOAPEnvelope()
161        requestEnvelope.create()
162        request = requestEnvelope.serialize()
163       
164        response = self.app.post('/my-soap-endpoint', 
165                                 params=request, 
166                                 status=200)
167        print(response.headers)
168        print(response.status)
169        print(response.body)
170
171    def test02Urllib2Client(self):
172       
173        # Paster based service is threaded from this call
174        self.addService(app=SOAPBindingMiddleware(), 
175                        port=self.__class__.SOAP_SERVICE_PORTNUM)
176       
177        client = UrlLib2SOAPClient()
178       
179        # ElementTree based envelope class
180        client.responseEnvelopeClass = SOAPEnvelope
181       
182        request = UrlLib2SOAPRequest()
183        request.url = self.__class__.ENDPOINT
184        request.envelope = SOAPEnvelope()
185        request.envelope.create()
186       
187        client.openerDirector.add_handler(HTTPHandler())
188        try:
189            response = client.send(request)
190        except URLError, e:
191            self.fail("soap_server.py must be running for this test")
192       
193        print("Response from server:\n\n%s" % response.envelope.serialize())
194       
195    def addService(self, *arg, **kw):
196        """Utility for setting up threads to run Paste HTTP based services with
197        unit tests
198       
199        @param arg: tuple contains ini file path setting for the service
200        @type arg: tuple
201        @param kw: keywords including "port" - port number to run the service
202        from
203        @type kw: dict
204        """
205        if self.disableServiceStartup:
206            return
207       
208        try:
209            self.services.append(PasteDeployAppServer(*arg, **kw))
210            self.services[-1].startThread()
211           
212        except socket.error:
213            pass
214
215    def __del__(self):
216        """Stop any services started with the addService method and clean up
217        the CA directory following the trust roots call
218        """
219        if hasattr(self, 'services'):
220            for service in self.services:
221                service.terminateThread()
222
223
224if __name__ == "__main__":
225    unittest.main()
Note: See TracBrowser for help on using the repository browser.