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

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

Merged ndg.soap back into ndg_saml package - simpler to keep it together.

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.etree import SOAPEnvelope
24from ndg.soap.client import UrlLib2SOAPClient, UrlLib2SOAPRequest
25from ndg.soap.test import PasteDeployAppServer
26
27
28class SOAPBindingMiddleware(object):
29    """Simple WSGI interface for SOAP service"""
30       
31    def __call__(self, environ, start_response):
32        requestFile = environ['wsgi.input']
33       
34        print("Server received request from client:\n\n%s" % 
35              requestFile.read())
36       
37        soapResponse = SOAPEnvelope()
38        soapResponse.create()
39       
40        response = soapResponse.serialize()
41        start_response("200 OK",
42                       [('Content-length', str(len(response))),
43                        ('Content-type', 'text/xml')])
44        return [response]
45   
46   
47class SOAPTestCase(unittest.TestCase):
48    SOAP_SERVICE_PORTNUM = 10080
49    ENDPOINT = 'http://localhost:%d/soap' % SOAP_SERVICE_PORTNUM
50    THIS_DIR = path.abspath(path.dirname(__file__))
51   
52    def __init__(self, *args, **kwargs):
53        """Use paste.fixture to test client/server SOAP interface"""
54        self.services = []
55        self.disableServiceStartup = False
56
57        wsgiApp = SOAPBindingMiddleware()
58        self.app = paste.fixture.TestApp(wsgiApp)
59         
60        super(SOAPTestCase, self).__init__(*args, **kwargs)
61       
62    def test01Envelope(self):
63        envelope = SOAPEnvelope()
64        envelope.create()
65        soap = envelope.serialize()
66       
67        self.assert_(len(soap) > 0)
68        self.assert_("Envelope" in soap)
69        self.assert_("Body" in soap)
70        self.assert_("Header" in soap)
71       
72        print(envelope.prettyPrint())
73        stream = StringIO()
74        stream.write(soap)
75        stream.seek(0)
76       
77        envelope2 = SOAPEnvelope()
78        envelope2.parse(stream)
79        soap2 = envelope2.serialize()
80        self.assert_(soap2 == soap)
81
82    def test02SendRequest(self):
83        requestEnvelope = SOAPEnvelope()
84        requestEnvelope.create()
85        request = requestEnvelope.serialize()
86       
87        response = self.app.post('/my-soap-endpoint', 
88                                 params=request, 
89                                 status=200)
90        print(response.headers)
91        print(response.status)
92        print(response.body)
93
94    def test03Urllib2Client(self):
95       
96        # Paster based service is threaded from this call
97        self.addService(app=SOAPBindingMiddleware(), 
98                        port=SOAPTestCase.SOAP_SERVICE_PORTNUM)
99       
100        client = UrlLib2SOAPClient()
101       
102        # ElementTree based envelope class
103        client.responseEnvelopeClass = SOAPEnvelope
104       
105        request = UrlLib2SOAPRequest()
106        request.url = SOAPTestCase.ENDPOINT
107        request.envelope = SOAPEnvelope()
108        request.envelope.create()
109       
110        client.openerDirector.add_handler(HTTPHandler())
111        try:
112            response = client.send(request)
113        except URLError, e:
114            self.fail("soap_server.py must be running for this test")
115       
116        print("Response from server:\n\n%s" % response.envelope.serialize())
117       
118    def addService(self, *arg, **kw):
119        """Utility for setting up threads to run Paste HTTP based services with
120        unit tests
121       
122        @param arg: tuple contains ini file path setting for the service
123        @type arg: tuple
124        @param kw: keywords including "port" - port number to run the service
125        from
126        @type kw: dict
127        """
128        if self.disableServiceStartup:
129            return
130       
131        try:
132            self.services.append(PasteDeployAppServer(*arg, **kw))
133            self.services[-1].startThread()
134           
135        except socket.error:
136            pass
137
138    def __del__(self):
139        """Stop any services started with the addService method and clean up
140        the CA directory following the trust roots call
141        """
142        if hasattr(self, 'services'):
143            for service in self.services:
144                service.terminateThread()
145
146
147if __name__ == "__main__":
148    unittest.main()
Note: See TracBrowser for help on using the repository browser.