source: TI12-security/trunk/WSSecurity/ndg/wssecurity/test/unit/__init__.py @ 6396

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/WSSecurity/ndg/wssecurity/test/unit/__init__.py@6396
Revision 6396, 4.1 KB checked in by pjkersha, 11 years ago (diff)
Line 
1"""NDG Security unit test package
2
3NERC DataGrid 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()
16log = logging.getLogger(__name__)
17
18import os
19from os.path import expandvars, join, dirname, abspath
20
21try:
22    from hashlib import md5
23except ImportError:
24    # Allow for < Python 2.5
25    from md5 import md5
26
27
28TEST_CONFIG_DIR = join(abspath(dirname(dirname(__file__))), 'config')
29
30mkDataDirPath = lambda file:join(TEST_CONFIG_DIR, file)
31
32from ndg.security.common.X509 import X500DN
33from ndg.security.test.unit.wsgi import PasteDeployAppServer
34
35try:
36    from sqlalchemy import (create_engine, MetaData, Table, Column, Integer, 
37                            String)
38    from sqlalchemy.ext.declarative import declarative_base
39    from sqlalchemy.orm import sessionmaker
40   
41    sqlAlchemyInstalled = True
42except ImportError:
43    sqlAlchemyInstalled = False
44   
45   
46class BaseTestCase(unittest.TestCase):
47    '''Convenience base class from which other unit tests can extend.  Its
48    sets the generic data directory path'''
49    TEST_CONFIG_DIR_VARNAME = 'NDGSEC_TEST_CONFIG_DIR'
50
51   
52    NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES_ENVVAR = \
53        'NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES'
54   
55    _disableServiceStartup = lambda self: bool(os.environ.get(
56        BaseTestCase.NDGSEC_UNITTESTS_DISABLE_THREAD_SERVICES_ENVVAR))
57   
58    disableServiceStartup = property(fget=_disableServiceStartup,
59                                     doc="Stop automated start-up of services "
60                                         "for unit tests")
61   
62    NDGSEC_TEST_CONFIG_DIR = os.environ.get(TEST_CONFIG_DIR_VARNAME, 
63                                            TEST_CONFIG_DIR)
64   
65    CACERT_DIR = os.path.join(NDGSEC_TEST_CONFIG_DIR, 'ca')
66    PKI_DIR = os.path.join(NDGSEC_TEST_CONFIG_DIR, 'pki')
67   
68    def __init__(self, *arg, **kw):
69        if BaseTestCase.TEST_CONFIG_DIR_VARNAME not in os.environ:
70            os.environ[BaseTestCase.TEST_CONFIG_DIR_VARNAME] = TEST_CONFIG_DIR
71               
72        unittest.TestCase.__init__(self, *arg, **kw)
73        self.services = []
74       
75    def addService(self, *arg, **kw):
76        """Utility for setting up threads to run Paste HTTP based services with
77        unit tests
78       
79        @param cfgFilePath: ini file containing configuration for the service
80        @type cfgFilePath: basestring
81        @param port: port number to run the service from
82        @type port: int
83        """
84        if self.disableServiceStartup:
85            return
86       
87        withSSL = kw.pop('withSSL', False)
88        if withSSL:
89            from OpenSSL import SSL
90           
91            certFilePath = mkDataDirPath(os.path.join('pki', 'localhost.crt'))
92            priKeyFilePath = mkDataDirPath(os.path.join('pki', 'localhost.key'))
93           
94            kw['ssl_context'] = SSL.Context(SSL.SSLv23_METHOD)
95            kw['ssl_context'].set_options(SSL.OP_NO_SSLv2)
96       
97            kw['ssl_context'].use_privatekey_file(priKeyFilePath)
98            kw['ssl_context'].use_certificate_file(certFilePath)
99           
100        try:
101            self.services.append(PasteDeployAppServer(*arg, **kw))
102            self.services[-1].startThread()
103           
104        except socket.error:
105            pass
106       
107    def __del__(self):
108        """Stop any services started with the addService method"""
109        if hasattr(self, 'services'):
110            for service in self.services:
111                service.terminateThread()
112
113def _getParentDir(depth=0, path=dirname(__file__)):
114    """
115    @type path: basestring
116    @param path: directory path from which to get parent directory, defaults
117    to dir of this module
118    @rtype: basestring
119    @return: parent directory at depth levels up from the current path
120    """
121    for i in range(depth):
122        path = dirname(path)
123    return path
124
125
Note: See TracBrowser for help on using the repository browser.