path: root/QMTest/scons_tdb.py

#!/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.


"""
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 list(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: