source: TI12-security/trunk/python/ndg.security.test/ndg/security/test/unit/wssecurity/foursuite/server/echoServer.py @ 5290

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/python/ndg.security.test/ndg/security/test/unit/wssecurity/foursuite/server/echoServer.py@5290
Revision 5290, 4.1 KB checked in by pjkersha, 10 years ago (diff)

Reworking unit tests

  • Property svn:executable set to *
Line 
1#!/usr/bin/env python
2#
3# How to build an echo server using the extended code generation
4#
5import sys
6import os
7from ConfigParser import SafeConfigParser
8
9# Import the ZSI stuff you'd need no matter what
10from ZSI.ServiceContainer import ServiceContainer
11
12# This is a new method imported to show it's value
13from ZSI.ServiceContainer import GetSOAPContext
14
15from EchoService_services_server import EchoService as _EchoService
16
17from ndg.security.common.wssecurity.signaturehandler.dom import SignatureHandler
18
19from os.path import expandvars as xpdVars
20from os.path import join, dirname, abspath
21mkPath = lambda file: join(os.environ['NDGSEC_WSSESRV_UNITTEST_DIR'], file)
22
23import logging
24log = logging.getLogger(__name__)
25logging.basicConfig(level=logging.DEBUG,
26                    format='%(asctime)s %(filename)s:%(lineno)d '
27                    '%(levelname)s %(message)s')
28
29
30from ndg.security.test.unit import BaseTestCase
31
32# Initialize environment for unit tests
33if BaseTestCase.configDirEnvVarName not in os.environ:
34    os.environ[BaseTestCase.configDirEnvVarName] = \
35                join(dirname(dirname(abspath(dirname(__file__)))), 'config')
36
37class EchoService(_EchoService):
38
39    def __init__(self, **kw):
40       
41        # Stop in debugger at beginning of SOAP stub if environment variable
42        # is set
43        self.__debug = bool(os.environ.get('NDGSEC_INT_DEBUG'))
44        if self.__debug:
45            import pdb
46            pdb.set_trace()
47           
48        _EchoService.__init__(self, **kw)
49       
50    def sign(self, sw):
51        '''\
52        Overrides ServiceInterface class method to allow digital signature'''
53        self.signatureHandler.sign(sw)
54       
55    def verify(self, ps):
56        '''\
57        Overrides ServiceInterface class method to allow signature
58        verification'''     
59        self.signatureHandler.verify(ps)
60       
61    def soap_Echo(self, ps, **kw):
62        '''Simple echo method to test WS-Security DSIG
63       
64        @type ps: ZSI ParsedSoap
65        @param ps: client SOAP message
66        @rtype: tuple
67        @return: response objects'''
68        log.info("Server received an Echo service request...")
69        if self.__debug:
70            import pdb
71            pdb.set_trace()
72       
73        response = _EchoService.soap_Echo(self, ps)   
74        response.EchoResult = "Received message from client: " + \
75                            self.request.EchoIn
76        return response
77   
78   
79    def authorize(self, auth_info, post, action):
80        '''Override default simply to display client request info'''
81        ctx = GetSOAPContext()
82        print "-"*80
83        print dir(ctx)
84        print "Container: ", ctx.connection
85        print "Parsed SOAP: ", ctx.parsedsoap
86        print "Container: ", ctx.container
87        print "HTTP Headers:\n", ctx.httpheaders
88        print "-"*80
89        print "Client Request:\n", ctx.xmldata
90        return 1
91
92   
93if __name__ == "__main__":
94    # Here we set up the server
95       
96    if 'NDGSEC_WSSESRV_UNITTEST_DIR' not in os.environ:
97        os.environ['NDGSEC_WSSESRV_UNITTEST_DIR'] = abspath(dirname(__file__))
98
99    if 'NDGSEC_TEST_CONFIG_DIR' not in os.environ:
100        os.environ['NDGSEC_TEST_CONFIG_DIR'] = \
101            abspath(join(dirname(dirname(dirname(dirname(__file__)))),
102                         'config'))
103           
104    configFilePath = mkPath('echoServer.cfg')
105
106    cfg = SafeConfigParser()
107    files = cfg.read(configFilePath)
108    assert len(files) == 1, "Error reading %s" % configFilePath
109   
110    hostname = cfg.get('setUp', 'hostname')
111    port = cfg.getint('setUp', 'port')
112    path = cfg.get('setUp', 'path')
113   
114    wsseCfgFilePath = xpdVars(cfg.get('setUp', 'wsseCfgFilePath'))
115
116    serviceContainer = ServiceContainer((hostname, port))   
117   
118    # Create the Inherited version of the server
119    echo = EchoService()
120    echo.signatureHandler = SignatureHandler(cfg=wsseCfgFilePath)
121
122    serviceContainer.setNode(echo, url=path)
123   
124    try:
125        # Run the service container
126        print "listening at http://%s:%s%s" % (hostname, port, path)
127        serviceContainer.serve_forever()
128    except KeyboardInterrupt:
129        sys.exit(0)
Note: See TracBrowser for help on using the repository browser.