summaryrefslogtreecommitdiffstats
path: root/QMTest/scons_tdb.py
diff options
context:
space:
mode:
Diffstat (limited to 'QMTest/scons_tdb.py')
-rw-r--r--QMTest/scons_tdb.py520
1 files changed, 520 insertions, 0 deletions
diff --git a/QMTest/scons_tdb.py b/QMTest/scons_tdb.py
new file mode 100644
index 0000000..145c2a7
--- /dev/null
+++ b/QMTest/scons_tdb.py
@@ -0,0 +1,520 @@
+#!/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.
+#
+
+__revision__ = "__FILE__ __REVISION__ __DATE__ __DEVELOPER__"
+
+"""
+QMTest classes to support SCons' testing and Aegis-inspired workflow.
+
+Thanks to Stefan Seefeld for the initial code.
+"""
+
+########################################################################
+# 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.directory_suite import DirectorySuite
+from qm.extension import get_extension_class_name, get_class_arguments_as_dictionary
+import os, dircache
+
+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 e.__dict__.has_key(name):
+ 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
+
+# XXX I'd like to annotate the overall test run with the following
+# information about the Python version, SCons version, and environment.
+# Not sure how to do that yet; ask Stefan.
+#
+# sys_keys = ['byteorder', 'exec_prefix', 'executable', 'maxint', 'maxunicode', 'platform', 'prefix', 'version', 'version_info']
+
+# " <%s>" % tag
+# " <version>%s</version>" % module.__version__
+# " <build>%s</build>" % module.__build__
+# " <buildsys>%s</buildsys>" % module.__buildsys__
+# " <date>%s</date>" % module.__date__
+# " <developer>%s</developer>" % module.__developer__
+# " </%s>" % tag
+
+# " <scons>"
+# print_version_info("script", scons)
+# print_version_info("engine", SCons)
+# " </scons>"
+
+# environ_keys = [
+# 'PATH',
+# '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',
+# 'LOGNAME',
+# 'MACHINE',
+# 'OLDPWD',
+# 'PWD',
+# 'OPSYS',
+# 'SHELL',
+# 'TMPDIR',
+# 'USER',
+# ]
+
+class AegisStream(TextResultStream):
+ 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 result["Test.print_time"] != "0":
+ 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):
+ arguments = [
+ qm.fields.TextField(
+ name = "results_file",
+ title = "Aegis Results File",
+ description = """
+ """,
+ verbatim = "true",
+ default_value = "aegis-results.txt",
+ ),
+ ]
+ def __init__(self, arguments):
+ self.filename = arguments['results_file']
+ 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, exit_status in self._outcomes.items():
+ 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', 'python'), self.script]
+ status = command.Run(args, os.environ)
+ result["Test.print_time"] = context.get('print_time', '0')
+ 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 = 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 = 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)