#!/usr/bin/env python
#
# __COPYRIGHT__
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY
# KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from __future__ import division
"""
QMTest classes to support SCons' testing and Aegis-inspired workflow.
Thanks to Stefan Seefeld for the initial code.
"""
__revision__ = "__FILE__ __REVISION__ __DATE__ __DEVELOPER__"
########################################################################
# Imports
########################################################################
import qm
import qm.common
import qm.test.base
from qm.fields import *
from qm.executable import *
from qm.test import database
from qm.test import test
from qm.test import resource
from qm.test import suite
from qm.test.result import Result
from qm.test.file_result_stream import FileResultStream
from qm.test.classes.text_result_stream import TextResultStream
from qm.test.classes.xml_result_stream import XMLResultStream
from qm.test.directory_suite import DirectorySuite
from qm.extension import get_extension_class_name, get_class_arguments_as_dictionary
import dircache
import os
import imp
if sys.platform == 'win32':
console = 'con'
else:
console = '/dev/tty'
def Trace(msg):
open(console, 'w').write(msg)
# QMTest 2.3 hard-codes how it captures the beginning and end time by
# calling the qm.common.format_time_iso() function, which canonicalizes
# the time stamp in one-second granularity ISO format. In order to get
# sub-second granularity, as well as to use the more precise time.clock()
# function on Windows, we must replace that function with our own.
orig_format_time_iso = qm.common.format_time_iso
if sys.platform == 'win32':
time_func = time.clock
else:
time_func = time.time
def my_format_time(time_secs=None):
return str(time_func())
qm.common.format_time_iso = my_format_time
########################################################################
# Classes
########################################################################
def get_explicit_arguments(e):
"""This function can be removed once QMTest 2.4 is out."""
# Get all of the arguments.
arguments = get_class_arguments_as_dictionary(e.__class__)
# Determine which subset of the 'arguments' have been set
# explicitly.
explicit_arguments = {}
for name, field in arguments.items():
# Do not record computed fields.
if field.IsComputed():
continue
if name in e.__dict__:
explicit_arguments[name] = e.__dict__[name]
return explicit_arguments
def check_exit_status(result, prefix, desc, status):
"""This function can be removed once QMTest 2.4 is out."""
if sys.platform == "win32" or os.WIFEXITED(status):
# Obtain the exit code.
if sys.platform == "win32":
exit_code = status
else:
exit_code = os.WEXITSTATUS(status)
# If the exit code is non-zero, the test fails.
if exit_code != 0:
result.Fail("%s failed with exit code %d." % (desc, exit_code))
# Record the exit code in the result.
result[prefix + "exit_code"] = str(exit_code)
return False
elif os.WIFSIGNALED(status):
# Obtain the signal number.
signal = os.WTERMSIG(status)
# If the program gets a fatal signal, the test fails .
result.Fail("%s received fatal signal %d." % (desc, signal))
result[prefix + "signal"] = str(signal)
return False
else:
# A process should only be able to stop by exiting, or
# by being terminated with a signal.
assert None
return True
class Null:
pass
_null = Null()
sys_attributes = [
'byteorder',
'exec_prefix',
'executable',
'maxint',
'maxunicode',
'platform',
'prefix',
'version',
'version_info',
]
def get_sys_values():
sys_attributes.sort()
result = [(k, getattr(sys, k, _null)) for k in sys_attributes]
result = [t for t in result if not t[1] is _null]
result = [t[0] + '=' + repr(t[1]) for t in result]
return '\n '.join(result)
module_attributes = [
'__version__',
'__build__',
'__buildsys__',
'__date__',
'__developer__',
]
def get_module_info(module):
module_attributes.sort()
result = [(k, getattr(module, k, _null)) for k in module_attributes]
result = [t for t in result if not t[1] is _null]
result = [t[0] + '=' + repr(t[1]) for t in result]
return '\n '.join(result)
environ_keys = [
'PATH',
'SCONS',
'SCONSFLAGS',
'SCONS_LIB_DIR',
'PYTHON_ROOT',
'QTDIR',
'COMSPEC',
'INTEL_LICENSE_FILE',
'INCLUDE',
'LIB',
'MSDEVDIR',
'OS',
'PATHEXT',
'SystemRoot',
'TEMP',
'TMP',
'USERNAME',
'VXDOMNTOOLS',
'WINDIR',
'XYZZY'
'ENV',
'HOME',
'LANG',
'LANGUAGE',
'LC_ALL',
'LC_MESSAGES',
'LOGNAME',
'MACHINE',
'OLDPWD',
'PWD',
'OPSYS',
'SHELL',
'TMPDIR',
'USER',
]
def get_environment():
environ_keys.sort()
result = [(k, os.environ.get(k, _null)) for k in environ_keys]
result = [t for t in result if not t[1] is _null]
result = [t[0] + '-' + t[1] for t in result]
return '\n '.join(result)
class SConsXMLResultStream(XMLResultStream):
def __init__(self, *args, **kw):
super(SConsXMLResultStream, self).__init__(*args, **kw)
def WriteAllAnnotations(self, context):
# Load (by hand) the SCons modules we just unwrapped so we can
# extract their version information. Note that we have to override
# SCons.Script.main() with a do_nothing() function, because loading up
# the 'scons' script will actually try to execute SCons...
src_engine = os.environ.get('SCONS_LIB_DIR')
if not src_engine:
src_engine = os.path.join('src', 'engine')
fp, pname, desc = imp.find_module('SCons', [src_engine])
SCons = imp.load_module('SCons', fp, pname, desc)
# Override SCons.Script.main() with a do-nothing function, because
# loading the 'scons' script will actually try to execute SCons...
src_engine_SCons = os.path.join(src_engine, 'SCons')
fp, pname, desc = imp.find_module('Script', [src_engine_SCons])
SCons.Script = imp.load_module('Script', fp, pname, desc)
def do_nothing():
pass
SCons.Script.main = do_nothing
scons_file = os.environ.get('SCONS')
if scons_file:
src_script, scons_py = os.path.split(scons_file)
scons = os.path.splitext(scons_py)[0]
else:
src_script = os.path.join('src', 'script')
scons = 'scons'
fp, pname, desc = imp.find_module(scons, [src_script])
scons = imp.load_module('scons', fp, pname, desc)
fp.close()
self.WriteAnnotation("scons_test.engine", get_module_info(SCons))
self.WriteAnnotation("scons_test.script", get_module_info(scons))
self.WriteAnnotation("scons_test.sys", get_sys_values())
self.WriteAnnotation("scons_test.os.environ", get_environment())
class AegisStream(TextResultStream):
arguments = [
qm.fields.IntegerField(
name = "print_time",
title = "print individual test times",
description = """
""",
default_value = 0,
),
]
def __init__(self, *args, **kw):
super(AegisStream, self).__init__(*args, **kw)
self._num_tests = 0
self._outcomes = {}
self._outcome_counts = {}
for outcome in AegisTest.aegis_outcomes:
self._outcome_counts[outcome] = 0
self.format = "full"
def _percent(self, outcome):
return 100. * self._outcome_counts[outcome] / self._num_tests
def _aegis_no_result(self, result):
outcome = result.GetOutcome()
return (outcome == Result.FAIL and result.get('Test.exit_code') == '2')
def _DisplayText(self, text):
# qm.common.html_to_text() uses htmllib, which sticks an extra
# '\n' on the front of the text. Strip it and only display
# the text if there's anything to display.
text = qm.common.html_to_text(text)
if text[0] == '\n':
text = text[1:]
if text:
lines = text.splitlines()
if lines[-1] == '':
lines = lines[:-1]
self.file.write(' ' + '\n '.join(lines) + '\n\n')
def _DisplayResult(self, result, format):
test_id = result.GetId()
kind = result.GetKind()
if self._aegis_no_result(result):
outcome = "NO_RESULT"
else:
outcome = result.GetOutcome()
self._WriteOutcome(test_id, kind, outcome)
self.file.write('\n')
def _DisplayAnnotations(self, result):
try:
self._DisplayText(result["Test.stdout"])
except KeyError:
pass
try:
self._DisplayText(result["Test.stderr"])
except KeyError:
pass
if self.print_time:
start = float(result['qmtest.start_time'])
end = float(result['qmtest.end_time'])
fmt = " Total execution time: %.1f seconds\n\n"
self.file.write(fmt % (end - start))
class AegisChangeStream(AegisStream):
def WriteResult(self, result):
test_id = result.GetId()
if self._aegis_no_result(result):
outcome = AegisTest.NO_RESULT
else:
outcome = result.GetOutcome()
self._num_tests += 1
self._outcome_counts[outcome] += 1
super(AegisStream, self).WriteResult(result)
def _SummarizeTestStats(self):
self.file.write("\n")
self._DisplayHeading("STATISTICS")
if self._num_tests != 0:
# We'd like to use the _FormatStatistics() method to do
# this, but it's wrapped around the list in Result.outcomes,
# so it's simpler to just do it ourselves.
print " %6d tests total\n" % self._num_tests
for outcome in AegisTest.aegis_outcomes:
if self._outcome_counts[outcome] != 0:
print " %6d (%3.0f%%) tests %s" % (
self._outcome_counts[outcome],
self._percent(outcome),
outcome
)
class AegisBaselineStream(AegisStream):
def WriteResult(self, result):
test_id = result.GetId()
if self._aegis_no_result(result):
outcome = AegisTest.NO_RESULT
self.expected_outcomes[test_id] = Result.PASS
self._outcome_counts[outcome] += 1
else:
self.expected_outcomes[test_id] = Result.FAIL
outcome = result.GetOutcome()
if outcome != Result.Fail:
self._outcome_counts[outcome] += 1
self._num_tests += 1
super(AegisStream, self).WriteResult(result)
def _SummarizeRelativeTestStats(self):
self.file.write("\n")
self._DisplayHeading("STATISTICS")
if self._num_tests != 0:
# We'd like to use the _FormatStatistics() method to do
# this, but it's wrapped around the list in Result.outcomes,
# so it's simpler to just do it ourselves.
if self._outcome_counts[AegisTest.FAIL]:
print " %6d (%3.0f%%) tests as expected" % (
self._outcome_counts[AegisTest.FAIL],
self._percent(AegisTest.FAIL),
)
non_fail_outcomes = list(AegisTest.aegis_outcomes[:])
non_fail_outcomes.remove(AegisTest.FAIL)
for outcome in non_fail_outcomes:
if self._outcome_counts[outcome] != 0:
print " %6d (%3.0f%%) tests unexpected %s" % (
self._outcome_counts[outcome],
self._percent(outcome),
outcome,
)
class AegisBatchStream(FileResultStream):
def __init__(self, arguments):
super(AegisBatchStream, self).__init__(arguments)
self._outcomes = {}
def WriteResult(self, result):
test_id = result.GetId()
kind = result.GetKind()
outcome = result.GetOutcome()
exit_status = '0'
if outcome == Result.FAIL:
exit_status = result.get('Test.exit_code')
self._outcomes[test_id] = exit_status
def Summarize(self):
self.file.write('test_result = [\n')
for file_name in sorted(self._outcomes.keys()):
exit_status = self._outcomes[file_name]
file_name = file_name.replace('\\', '/')
self.file.write(' { file_name = "%s";\n' % file_name)
self.file.write(' exit_status = %s; },\n' % exit_status)
self.file.write('];\n')
class AegisTest(test.Test):
PASS = "PASS"
FAIL = "FAIL"
NO_RESULT = "NO_RESULT"
ERROR = "ERROR"
UNTESTED = "UNTESTED"
aegis_outcomes = (
PASS, FAIL, NO_RESULT, ERROR, UNTESTED,
)
"""Aegis test outcomes."""
class Test(AegisTest):
"""Simple test that runs a python script and checks the status
to determine whether the test passes."""
script = TextField(title="Script to test")
topdir = TextField(title="Top source directory")
def Run(self, context, result):
"""Run the test. The test passes if the command exits with status=0,
and fails otherwise. The program output is logged, but not validated."""
command = RedirectedExecutable()
args = [context.get('python', sys.executable), '-tt', self.script]
status = command.Run(args, os.environ)
if not check_exit_status(result, 'Test.', self.script, status):
# In case of failure record exit code, stdout, and stderr.
result.Fail("Non-zero exit_code.")
result["Test.stdout"] = result.Quote(command.stdout)
result["Test.stderr"] = result.Quote(command.stderr)
class Database(database.Database):
"""Scons test database.
* The 'src' and 'test' directories are explicit suites.
* Their subdirectories are implicit suites.
* All files under 'src/' ending with 'Tests.py' contain tests.
* All files under 'test/' with extension '.py' contain tests.
* Right now there is only a single test class, which simply runs
the specified python interpreter on the given script. To be refined..."""
srcdir = TextField(title = "Source Directory",
description = "The root of the test suite's source tree.")
_is_generic_database = True
def is_a_test_under_test(path, t):
return os.path.splitext(t)[1] == '.py' \
and os.path.isfile(os.path.join(path, t))
def is_a_test_under_src(path, t):
return t[-8:] == 'Tests.py' \
and os.path.isfile(os.path.join(path, t))
is_a_test = {
'src' : is_a_test_under_src,
'test' : is_a_test_under_test,
}
exclude_subdirs = {
'.svn' : 1,
'CVS' : 1,
}
def is_a_test_subdir(path, subdir):
if exclude_subdirs.get(subdir):
return None
return os.path.isdir(os.path.join(path, subdir))
def __init__(self, path, arguments):
self.label_class = "file_label.FileLabel"
self.modifiable = "false"
# Initialize the base class.
super(Database, self).__init__(path, arguments)
def GetRoot(self):
return self.srcdir
def GetSubdirectories(self, directory):
components = self.GetLabelComponents(directory)
path = os.path.join(self.GetRoot(), *components)
if directory:
dirs = [d for d in dircache.listdir(path)
if os.path.isdir(os.path.join(path, d))]
else:
dirs = list(self.is_a_test.keys())
dirs.sort()
return dirs
def GetIds(self, kind, directory = "", scan_subdirs = 1):
components = self.GetLabelComponents(directory)
path = os.path.join(self.GetRoot(), *components)
if kind == database.Database.TEST:
if not components:
return []
ids = [self.JoinLabels(directory, t)
for t in dircache.listdir(path)
if self.is_a_test[components[0]](path, t)]
elif kind == Database.RESOURCE:
return [] # no resources yet
else: # SUITE
if directory:
ids = [self.JoinLabels(directory, d)
for d in dircache.listdir(path)
if os.path.isdir(os.path.join(path, d))]
else:
ids = list(self.is_a_test.keys())
if scan_subdirs:
for d in dircache.listdir(path):
if (os.path.isdir(d)):
ids.extend(self.GetIds(kind,
self.JoinLabels(directory, d),
True))
return ids
def GetExtension(self, id):
if not id:
return DirectorySuite(self, id)
components = self.GetLabelComponents(id)
path = os.path.join(self.GetRoot(), *components)
if os.path.isdir(path): # a directory
return DirectorySuite(self, id)
elif os.path.isfile(path): # a test
arguments = {}
arguments['script'] = path
arguments['topdir'] = self.GetRoot()
return Test(arguments, qmtest_id = id, qmtest_database = self)
else: # nothing else to offer
return None
def GetTest(self, test_id):
"""This method can be removed once QMTest 2.4 is out."""
t = self.GetExtension(test_id)
if isinstance(t, test.Test):
return database.TestDescriptor(self,
test_id,
get_extension_class_name(t.__class__),
get_explicit_arguments(t))
raise database.NoSuchTestError(test_id)
def GetSuite(self, suite_id):
"""This method can be removed once QMTest 2.4 is out."""
if suite_id == "":
return DirectorySuite(self, "")
s = self.GetExtension(suite_id)
if isinstance(s, suite.Suite):
return s
raise database.NoSuchSuiteError(suite_id)
def GetResource(self, resource_id):
"""This method can be removed once QMTest 2.4 is out."""
r = self.GetExtension(resource_id)
if isinstance(r, resource.Resource):
return ResourceDescriptor(self,
resource_id,
get_extension_class_name(r.__class__),
get_explicit_arguments(r))
raise database.NoSuchResourceError(resource_id)
# Local Variables:
# tab-width:4
# indent-tabs-mode:nil
# End:
# vim: set expandtab tabstop=4 shiftwidth=4: