source: TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/__init__.py @ 5907

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/__init__.py@5907
Revision 5907, 4.2 KB checked in by pjkersha, 10 years ago (diff)

Move MyProxy? certificate callout application into ndg.security.server and ndg.security.test. Too much trouble as separate egg.

Line 
1"""NDG Security unit test package
2
3NERC Data Grid Project
4"""
5__author__ = "P J Kershaw"
6__date__ = "14/05/09"
7__copyright__ = "(C) 2009 Science and Technology Facilities Council"
8__license__ = "BSD - see LICENSE file in top-level directory"
9__contact__ = "Philip.Kershaw@stfc.ac.uk"
10__revision__ = '$Id: __init__.py 4840 2009-01-19 13:59:08Z pjkersha $'
11
12import unittest
13import logging
14import socket
15logging.basicConfig()
16
17import os
18from os.path import expandvars, join, dirname, abspath
19
20TEST_CONFIG_DIR = join(abspath(dirname(dirname(__file__))), 'config')
21
22mkDataDirPath = lambda file:join(TEST_CONFIG_DIR, file)
23
24from ndg.security.test.unit.wsgi import PasteDeployAppServer
25
26class BaseTestCase(unittest.TestCase):
27    '''Convenience base class from which other unit tests can extend.  Its
28    sets the generic data directory path'''
29    configDirEnvVarName = 'NDGSEC_TEST_CONFIG_DIR'
30    SITEA_ATTRIBUTEAUTHORITY_PORTNUM = 5000
31    SITEB_ATTRIBUTEAUTHORITY_PORTNUM = 5100
32    SESSIONMANAGER_PORTNUM = 5500
33   
34    mkDataDirPath = staticmethod(mkDataDirPath)
35   
36    NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES_ENVVAR = \
37        'NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES'
38   
39    _disableServiceStartup = lambda self: bool(os.environ.get(
40        BaseTestCase.NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES_ENVVAR))
41   
42    disableServiceStartup = property(fget=_disableServiceStartup,
43                                     doc="Stop automated start-up of services "
44                                         "for unit tests")
45   
46    def __init__(self, *arg, **kw):
47        if BaseTestCase.configDirEnvVarName not in os.environ:
48            os.environ[BaseTestCase.configDirEnvVarName] = TEST_CONFIG_DIR
49               
50        unittest.TestCase.__init__(self, *arg, **kw)
51        self.services = []
52       
53    def addService(self, *arg, **kw):
54        """Utility for setting up threads to run Paste HTTP based services with
55        unit tests
56       
57        @param cfgFilePath: ini file containing configuration for the service
58        @type cfgFilePath: basestring
59        @param port: port number to run the service from
60        @type port: int
61        """
62        if self.disableServiceStartup:
63            return
64       
65        try:
66            self.services.append(PasteDeployAppServer(*arg, **kw))
67            self.services[-1].startThread()
68           
69        except socket.error:
70            pass
71
72    def startAttributeAuthorities(self):
73        """Serve test Attribute Authorities to test against"""
74        self.startSiteAAttributeAuthority()
75        self.startSiteBAttributeAuthority()
76       
77    def startSiteAAttributeAuthority(self):
78        siteACfgFilePath = mkDataDirPath(join('attributeauthority', 
79                                              'sitea', 
80                                              'site-a.ini'))
81        self.addService(cfgFilePath=siteACfgFilePath, 
82                        port=BaseTestCase.SITEA_ATTRIBUTEAUTHORITY_PORTNUM)
83       
84    def startSiteBAttributeAuthority(self):
85        siteBCfgFilePath = mkDataDirPath(join('attributeauthority',
86                                              'siteb', 
87                                              'site-b.ini'))
88        self.addService(cfgFilePath=siteBCfgFilePath, 
89                        port=BaseTestCase.SITEB_ATTRIBUTEAUTHORITY_PORTNUM)       
90
91    def startSessionManager(self):
92        """Serve test Session Manager service"""
93        cfgFilePath = mkDataDirPath(join('sessionmanager', 
94                                         'session-manager.ini'))
95        self.addService(cfgFilePath=cfgFilePath, 
96                        port=BaseTestCase.SESSIONMANAGER_PORTNUM)
97       
98
99    def __del__(self):
100        """Stop any services started with the addService method"""
101        if hasattr(self, 'services'):
102            for service in self.services:
103                service.terminateThread()
104           
105def _getParentDir(depth=0, path=dirname(__file__)):
106    """
107    @type path: basestring
108    @param path: directory path from which to get parent directory, defaults
109    to dir of this module
110    @rtype: basestring
111    @return: parent directory at depth levels up from the current path
112    """
113    for i in range(depth):
114        path = dirname(path)
115    return path
116
117
Note: See TracBrowser for help on using the repository browser.