#!/usr/bin/env python
############################################################################
# Joshua R. Boverhof, LBNL
# See LBNLCopyright for copyright notice!
###########################################################################
from compiler.ast import Module
import StringIO, copy, getopt
import os, sys, unittest, urlparse, signal, time, warnings, subprocess
from ConfigParser import ConfigParser, NoSectionError, NoOptionError
from ZSI.wstools.TimeoutSocket import TimeoutError
"""Global Variables:
CONFIG_FILE -- configuration file
CONFIG_PARSER -- ConfigParser instance
DOCUMENT -- test section variable, specifying document style.
LITERAL -- test section variable, specifying literal encodings.
BROKE -- test section variable, specifying broken test.
TESTS -- test section variable, whitespace separated list of modules.
SECTION_CONFIGURATION -- configuration section, turn on/off debuggging.
TRACEFILE -- file class instance.
TOPDIR -- current working directory
MODULEDIR -- stubs directory
PORT -- port of local container
HOST -- address of local container
SECTION_SERVERS -- services to be tested, values are paths to executables.
"""
CONFIG_FILE = 'config.txt'
CONFIG_PARSER = ConfigParser()
DOCUMENT = 'document'
LITERAL = 'literal'
BROKE = 'broke'
TESTS = 'tests'
SECTION_CONFIGURATION = 'configuration'
SECTION_DISPATCH = 'dispatch'
TRACEFILE = sys.stdout
TOPDIR = os.getcwd()
MODULEDIR = TOPDIR + '/stubs'
SECTION_SERVERS = 'servers'
CONFIG_PARSER.read(CONFIG_FILE)
DEBUG = CONFIG_PARSER.getboolean(SECTION_CONFIGURATION, 'debug')
SKIP = CONFIG_PARSER.getboolean(SECTION_CONFIGURATION, 'skip')
TWISTED = CONFIG_PARSER.getboolean(SECTION_CONFIGURATION, 'twisted')
LAZY = CONFIG_PARSER.getboolean(SECTION_CONFIGURATION, 'lazy')
OUTPUT = CONFIG_PARSER.get(SECTION_CONFIGURATION, 'output') or sys.stdout
if DEBUG:
from ZSI.wstools.logging import setBasicLoggerDEBUG
setBasicLoggerDEBUG()
sys.path.append('%s/%s' %(os.getcwd(), 'stubs'))
ENVIRON = copy.copy(os.environ)
ENVIRON['PYTHONPATH'] = ENVIRON.get('PYTHONPATH', '') + ':' + MODULEDIR
def _SimpleMain():
"""Gets tests to run from configuration file.
"""
unittest.TestProgram(defaultTest="all")
main = _SimpleMain
def _TwistedMain():
"""Gets tests to run from configuration file.
"""
from twisted.internet import reactor
reactor.callWhenRunning(_TwistedTestProgram, defaultTest="all")
reactor.run(installSignalHandlers=0)
if TWISTED: main = _TwistedMain
def _LaunchContainer(cmd):
'''
Parameters:
cmd -- executable, sets up a ServiceContainer or ?
'''
host = CONFIG_PARSER.get(SECTION_DISPATCH, 'host')
port = CONFIG_PARSER.get(SECTION_DISPATCH, 'port')
process = subprocess.Popen([cmd, port], env=ENVIRON)
time.sleep(1)
return process
class _TwistedTestProgram(unittest.TestProgram):
def runTests(self):
from twisted.internet import reactor
if self.testRunner is None:
self.testRunner = unittest.TextTestRunner(verbosity=self.verbosity)
result = self.testRunner.run(self.test)
reactor.stop()
return result.wasSuccessful()
class ConfigException(Exception):
"""Exception thrown when configuration settings arent correct.
"""
pass
class TestException(Exception):
"""Exception thrown when test case isn't correctly set up.
"""
pass
class ServiceTestCase(unittest.TestCase):
"""Conventions for method names:
test_net*
-- network tests
test_local*
-- local tests
test_dispatch*
-- tests that use the a spawned local container
class attributes: Edit/Override these in the inheriting class as needed
out -- file descriptor to write output to
name -- configuration item, must be set in class.
url_section -- configuration section, maps a test module
name to an URL.
client_file_name --
types_file_name --
server_file_name --
"""
out = OUTPUT
name = None
url_section = 'WSDL'
client_file_name = None
types_file_name = None
server_file_name = None
def __init__(self, methodName):
"""
parameters:
methodName --
instance variables:
client_module
types_module
server_module
processID
done
"""
self.methodName = methodName
self.url = None
self.wsdl2py_args = []
self.wsdl2dispatch_args = []
self.portkwargs = {}
self.client_module = self.types_module = self.server_module = None
self.done = False
if TWISTED:
self.wsdl2py_args.append('--twisted')
if LAZY:
self.wsdl2py_args.append('--lazy')
unittest.TestCase.__init__(self, methodName)
write = lambda self, arg: self.out.write(arg)
if sys.version_info[:2] >= (2,5):
_exc_info = unittest.TestCase._exc_info
else:
_exc_info = unittest.TestCase._TestCase__exc_info
def __call__(self, *args, **kwds):
self.run(*args, **kwds)
def run(self, result=None):
if result is None: result = self.defaultTestResult()
result.startTest(self)
testMethod = getattr(self, self.methodName)
try:
try:
self.setUp()
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
return
ok = False
try:
t1 = time.time()
pyobj = testMethod()
t2 = time.time()
ok = True
except self.failureException:
result.addFailure(self, self._exc_info())
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
try:
self.tearDown()
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
ok = False
if ok:
result.addSuccess(self)
print>>self
print>>self, "|"+"-"*60
print>>self, "| TestCase: %s" %self.methodName
print>>self, "|"+"-"*20
print>>self, "| run time: %s ms" %((t2-t1)*1000)
print>>self, "| return : %s" %pyobj
print>>self, "|"+"-"*60
finally:
result.stopTest(self)
def getPortKWArgs(self):
kw = {}
if CONFIG_PARSER.getboolean(SECTION_CONFIGURATION, 'tracefile'):
kw['tracefile'] = TRACEFILE
kw.update(self.portkwargs)
return kw
def _setUpDispatch(self):
"""Set this test up as a dispatch test.
url --
"""
host = CONFIG_PARSER.get(SECTION_DISPATCH, 'host')
port = CONFIG_PARSER.get(SECTION_DISPATCH, 'port')
path = CONFIG_PARSER.get(SECTION_DISPATCH, 'path')
scheme = 'http'
netloc = '%s:%s' %(host, port)
params = query = fragment = None
self.portkwargs['url'] = \
urlparse.urlunparse((scheme,netloc,path,params,query,fragment))
_wsdl = {}
def _generate(self):
"""call the wsdl2py and wsdl2dispatch scripts and
automatically add the "-f" or "-u" argument. Other args
can be appended via the "wsdl2py_args" and "wsdl2dispatch_args"
instance attributes.
"""
url = self.url
if SKIP:
ServiceTestCase._wsdl[url] = True
return
args = []
ServiceTestCase._wsdl[url] = False
if os.path.isfile(url):
args += ['-f', os.path.abspath(url)]
else:
args += ['-u', url]
try:
os.mkdir(MODULEDIR)
except OSError, ex:
pass
os.chdir(MODULEDIR)
if MODULEDIR not in sys.path:
sys.path.append(MODULEDIR)
try:
# Client Stubs
wsdl2py = ['wsdl2py'] + args + self.wsdl2py_args
try:
exit = subprocess.call(wsdl2py)
except OSError, ex:
warnings.warn("TODO: Not sure what is going on here?")
exit = -1
#TODO: returncode WINDOWS?
self.failUnless(os.WIFEXITED(exit),
'"%s" exited with signal#: %d' %(wsdl2py, exit))
self.failUnless(exit == 0,
'"%s" exited with exit status: %d' %(wsdl2py, exit))
# Service Stubs
if '-x' not in self.wsdl2py_args:
wsdl2dispatch = (['wsdl2dispatch'] + args +
self.wsdl2dispatch_args)
try:
exit = subprocess.call(wsdl2dispatch)
except OSError, ex:
warnings.warn("TODO: Not sure what is going on here?")
#TODO: returncode WINDOWS?
self.failUnless(os.WIFEXITED(exit),
'"%s" exited with signal#: %d' %(wsdl2dispatch, exit))
self.failUnless(exit == 0,
'"%s" exited with exit status: %d' %(wsdl2dispatch, exit))
ServiceTestCase._wsdl[url] = True
finally:
os.chdir(TOPDIR)
_process = None
_lastToDispatch = None
def setUp(self):
"""Generate types and services modules once, then make them
available thru the *_module attributes if the *_file_name
attributes were specified.
"""
section = self.url_section
name = self.name
if not section or not name:
raise TestException, 'section(%s) or name(%s) not defined' %(
section, name)
if not CONFIG_PARSER.has_section(section):
raise TestException,\
'No such section(%s) in configuration file(%s)' %(
self.url_section, CONFIG_FILE)
self.url = CONFIG_PARSER.get(section, name)
status = ServiceTestCase._wsdl.get(self.url)
if status is False:
self.fail('generation failed for "%s"' %self.url)
if status is None:
self._generate()
# Check for files
tfn = self.types_file_name
cfn = self.client_file_name
sfn = self.server_file_name
files = filter(lambda f: f is not None, [cfn, tfn,sfn])
if None is cfn is tfn is sfn:
return
for n,m in map(lambda i: (i,__import__(i.split('.py')[0])), files):
if tfn is not None and tfn == n:
self.types_module = m
elif cfn is not None and cfn == n:
self.client_module = m
elif sfn is not None and sfn == n:
self.server_module = m
else:
self.fail('Unexpected module %s' %n)
# DISPATCH PORTION OF SETUP
if not self.methodName.startswith('test_dispatch'):
return
self._setUpDispatch()
if ServiceTestCase._process is not None:
return
try:
expath = CONFIG_PARSER.get(SECTION_DISPATCH, name)
except (NoSectionError, NoOptionError), ex:
self.fail('section dispatch has no item "%s"' %name)
if ServiceTestCase._lastToDispatch == expath:
return
if ServiceTestCase._lastToDispatch is not None:
ServiceTestCase.CleanUp()
ServiceTestCase._lastToDispatch = expath
ServiceTestCase._process = _LaunchContainer(TOPDIR + '/' + expath)
def CleanUp(cls):
"""call this when dispatch server is no longer needed,
maybe another needs to be started. Assumption that
a single "Suite" uses the same server, once all the
tests are run in that suite do a cleanup.
"""
if cls._process is None:
return
os.kill(cls._process.pid, signal.SIGKILL)
cls._process = None
CleanUp = classmethod(CleanUp)
class ServiceTestSuite(unittest.TestSuite):
"""A test suite is a composite test consisting of a number of TestCases.
For use, create an instance of TestSuite, then add test case instances.
When all tests have been added, the suite can be passed to a test
runner, such as TextTestRunner. It will run the individual test cases
in the order in which they were added, aggregating the results. When
subclassing, do not forget to call the base class constructor.
"""
def __init__(self, tests=()):
unittest.TestSuite.__init__(self, tests)
def __call__(self, result):
# for python2.4
return self.run(result)
def addTest(self, test):
unittest.TestSuite.addTest(self, test)
def run(self, result):
for test in self._tests:
if result.shouldStop:
break
test(result)
ServiceTestCase.CleanUp()
return result
syntax highlighted by Code2HTML, v. 0.9.1