source: TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/soap/test_soap.py @ 7077

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/soap/test_soap.py@7077
Revision 7077, 3.5 KB checked in by pjkersha, 9 years ago (diff)
  • Property svn:keywords set to Id
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__ = "BSD - see LICENSE file in top-level directory"
11__contact__ = "Philip.Kershaw@stfc.ac.uk"
12__revision__ = "$Id$"
13import logging
14logging.basicConfig(level=logging.DEBUG)
15
16import unittest
17from cStringIO import StringIO
18import paste.fixture
19from urllib2 import HTTPHandler, URLError
20
21from ndg.security.test.unit import BaseTestCase
22from ndg.security.common.soap.etree import SOAPEnvelope
23from ndg.security.common.soap.client import UrlLib2SOAPClient, \
24    UrlLib2SOAPRequest
25
26class SOAPBindingMiddleware(object):
27    """Simple WSGI interface for SOAP service"""
28       
29    def __call__(self, environ, start_response):
30        requestFile = environ['wsgi.input']
31       
32        print("Server received request from client:\n\n%s" % 
33              requestFile.read())
34       
35        soapResponse = SOAPEnvelope()
36        soapResponse.create()
37       
38        response = soapResponse.serialize()
39        start_response("200 OK",
40                       [('Content-length', str(len(response))),
41                        ('Content-type', 'text/xml')])
42        return [response]
43   
44   
45class SOAPTestCase(BaseTestCase):
46    SOAP_SERVICE_PORTNUM = 10080
47    ENDPOINT = 'http://localhost:%d/soap' % SOAP_SERVICE_PORTNUM
48   
49    def __init__(self, *args, **kwargs):
50        """Use paste.fixture to test client/server SOAP interface"""
51        wsgiApp = SOAPBindingMiddleware()
52        self.app = paste.fixture.TestApp(wsgiApp)
53         
54        super(SOAPTestCase, self).__init__(*args, **kwargs)
55       
56    def test01Envelope(self):
57        envelope = SOAPEnvelope()
58        envelope.create()
59        soap = envelope.serialize()
60       
61        self.assert_(len(soap) > 0)
62        self.assert_("Envelope" in soap)
63        self.assert_("Body" in soap)
64        self.assert_("SOAP-ENV:Header" in soap)
65       
66        print(envelope.prettyPrint())
67        stream = StringIO()
68        stream.write(soap)
69        stream.seek(0)
70       
71        envelope2 = SOAPEnvelope()
72        envelope2.parse(stream)
73        soap2 = envelope2.serialize()
74        self.assert_(soap2 == soap)
75
76    def test02SendRequest(self):
77        requestEnvelope = SOAPEnvelope()
78        requestEnvelope.create()
79        request = requestEnvelope.serialize()
80       
81        response = self.app.post('/my-soap-endpoint', 
82                                 params=request, 
83                                 status=200)
84        print(response.headers)
85        print(response.status)
86        print(response.body)
87
88    def test03Urllib2Client(self):
89       
90        # Paster based service is threaded from this call
91        self.addService(app=SOAPBindingMiddleware(), 
92                        port=SOAPTestCase.SOAP_SERVICE_PORTNUM)
93       
94        client = UrlLib2SOAPClient()
95       
96        # ElementTree based envelope class
97        client.responseEnvelopeClass = SOAPEnvelope
98       
99        request = UrlLib2SOAPRequest()
100        request.url = SOAPTestCase.ENDPOINT
101        request.envelope = SOAPEnvelope()
102        request.envelope.create()
103       
104        client.openerDirector.add_handler(HTTPHandler())
105        try:
106            response = client.send(request)
107        except URLError, e:
108            self.fail("soap_server.py must be running for this test")
109       
110        print("Response from server:\n\n%s" % response.envelope.serialize())
111
112if __name__ == "__main__":
113    unittest.main()
Note: See TracBrowser for help on using the repository browser.