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

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

Integrated automated start-up and shutdown of Paste http servers for unit tests.

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        siteACfgFilePath = mkDataDirPath(join('attributeauthority', 
75                                              'sitea', 
76                                              'site-a.ini'))
77        self.addService(cfgFilePath=siteACfgFilePath, 
78                        port=BaseTestCase.SITEA_ATTRIBUTEAUTHORITY_PORTNUM)
79       
80        siteBCfgFilePath = mkDataDirPath(join('attributeauthority',
81                                              'siteb', 
82                                              'site-b.ini'))
83        self.addService(cfgFilePath=siteBCfgFilePath, 
84                        port=BaseTestCase.SITEB_ATTRIBUTEAUTHORITY_PORTNUM)       
85
86    def startSessionManager(self):
87        """Serve test Session Manager service"""
88        cfgFilePath = mkDataDirPath(join('sessionmanager', 
89                                         'session-manager.ini'))
90        self.addService(cfgFilePath=cfgFilePath, 
91                        port=BaseTestCase.SESSIONMANAGER_PORTNUM)
92       
93
94    def __del__(self):
95        """Stop any services started with the addService method"""
96        if hasattr(self, 'services'):
97            for service in self.services:
98                service.terminateThread()
99           
100def _getParentDir(depth=0, path=dirname(__file__)):
101    """
102    @type path: basestring
103    @param path: directory path from which to get parent directory, defaults
104    to dir of this module
105    @rtype: basestring
106    @return: parent directory at depth levels up from the current path
107    """
108    for i in range(depth):
109        path = dirname(path)
110    return path
111
112
Note: See TracBrowser for help on using the repository browser.