ext/testlib/handlers.py

# Copyright (c) 2017 Mark D. Hill and David A. Wood
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met: redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer;
# redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution;
# neither the name of the copyright holders nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Authors: Sean Wilson

'''
Handlers for the testlib Log.


'''
import multiprocessing
import os
import sys
import threading
import time
import traceback

import testlib.helper as helper
import testlib.log as log
import testlib.result as result
import testlib.state as state
import testlib.terminal as terminal

from queue import Queue, Empty
from testlib.configuration import constants


class _TestStreamManager(object):
    def __init__(self):
        self._writers = {}

    def open_writer(self, test_result):
        if test_result in self._writers:
            raise ValueError('Cannot have multiple writters on a single test.')
        self._writers[test_result] = _TestStreams(test_result.stdout,
                test_result.stderr)

    def get_writer(self, test_result):
        if test_result not in self._writers:
            self.open_writer(test_result)
        return self._writers[test_result]

    def close_writer(self, test_result):
        if test_result in self._writers:
            writer = self._writers.pop(test_result)
            writer.close()

    def close(self):
        for writer in self._writers.values():
            writer.close()
        self._writers.clear()

class _TestStreams(object):
    def __init__(self, stdout, stderr):
        helper.mkdir_p(os.path.dirname(stdout))
        helper.mkdir_p(os.path.dirname(stderr))
        self.stdout = open(stdout, 'w')
        self.stderr = open(stderr, 'w')

    def close(self):
        self.stdout.close()
        self.stderr.close()

class ResultHandler(object):
    '''
    Log handler which listens for test results and output saving data as
    it is reported.

    When the handler is closed it writes out test results in the python pickle
    format.
    '''
    def __init__(self, schedule, directory):
        '''
        :param schedule: The entire schedule as a :class:`LoadedLibrary`
            object.

        :param directory: Directory to save test stdout/stderr and aggregate
            results to.
        '''
        self.directory = directory
        self.internal_results = result.InternalLibraryResults(schedule,
                directory)
        self.test_stream_manager = _TestStreamManager()
        self._closed = False

        self.mapping = {
            log.LibraryStatus.type_id: self.handle_library_status,

            log.SuiteResult.type_id: self.handle_suite_result,
            log.TestResult.type_id: self.handle_test_result,

            log.TestStderr.type_id: self.handle_stderr,
            log.TestStdout.type_id: self.handle_stdout,
        }

    def handle(self, record):
        if not self._closed:
            self.mapping.get(record.type_id, lambda _:None)(record)

    def handle_library_status(self, record):
        if record['status'] in (state.Status.Complete, state.Status.Avoided):
            self.test_stream_manager.close()

    def handle_suite_result(self, record):
        suite_result = self.internal_results.get_suite_result(
                    record['metadata'].uid)
        suite_result.result = record['result']

    def handle_test_result(self, record):
        test_result = self._get_test_result(record)
        test_result.result = record['result']

    def handle_stderr(self, record):
        self.test_stream_manager.get_writer(
            self._get_test_result(record)
        ).stderr.write(record['buffer'])

    def handle_stdout(self, record):
        self.test_stream_manager.get_writer(
            self._get_test_result(record)
        ).stdout.write(record['buffer'])

    def _get_test_result(self, test_record):
        return self.internal_results.get_test_result(
                    test_record['metadata'].uid,
                    test_record['metadata'].suite_uid)

    def _save(self):
        #FIXME Hardcoded path name
        result.InternalSavedResults.save(
            self.internal_results,
            os.path.join(self.directory, constants.pickle_filename))
        result.JUnitSavedResults.save(
            self.internal_results,
            os.path.join(self.directory, constants.xml_filename))

    def close(self):
        if self._closed:
            return
        self._closed = True
        self._save()

    def unsuccessful(self):
        '''
        Performs an or reduce on all of the results.
        Returns true if at least one test is unsuccessful, false when all tests
        pass
        '''
        for suite_result in self.internal_results:
            if suite_result.unsuccessful:
                return True
        # If all are successful, then this wasn't "unsuccessful"
        return False


#TODO Change from a handler to an internal post processor so it can be used
# to reprint results
class SummaryHandler(object):
    '''
    A log handler which listens to the log for test results
    and reports the aggregate results when closed.
    '''
    color = terminal.get_termcap()
    reset = color.Normal
    colormap = {
            state.Result.Errored: color.Red,
            state.Result.Failed: color.Red,
            state.Result.Passed: color.Green,
            state.Result.Skipped: color.Cyan,
    }

    def __init__(self):
        self.mapping = {
            log.TestResult.type_id: self.handle_testresult,
            log.LibraryStatus.type_id: self.handle_library_status,
        }
        self._timer = helper.Timer()
        self.results = []

    def handle_library_status(self, record):
        if record['status'] == state.Status.Building:
            self._timer.restart()

    def handle_testresult(self, record):
        result = record['result'].value
        if result in (state.Result.Skipped, state.Result.Failed,
                state.Result.Passed, state.Result.Errored):
            self.results.append(result)

    def handle(self, record):
        self.mapping.get(record.type_id, lambda _:None)(record)

    def close(self):
        print(self._display_summary())

    def _display_summary(self):
        most_severe_outcome = None
        outcome_fmt = ' {count} {outcome}'
        strings = []

        outcome_count = [0] * len(state.Result.enums)
        for result in self.results:
            outcome_count[result] += 1

        # Iterate over enums so they are in order of severity
        for outcome in state.Result.enums:
            outcome = getattr(state.Result, outcome)
            count  = outcome_count[outcome]
            if count:
                strings.append(outcome_fmt.format(count=count,
                        outcome=state.Result.enums[outcome]))
                most_severe_outcome = outcome
        string = ','.join(strings)
        if most_severe_outcome is None:
            string = ' No testing done'
            most_severe_outcome = state.Result.Passed
        else:
            string = ' Results:' + string + ' in {:.2} seconds '.format(
                    self._timer.active_time())
        string += ' '
        return terminal.insert_separator(
                string,
                color=self.colormap[most_severe_outcome] + self.color.Bold)

class TerminalHandler(object):
    color = terminal.get_termcap()
    verbosity_mapping = {
        log.LogLevel.Warn: color.Yellow,
        log.LogLevel.Error: color.Red,
    }
    default = color.Normal

    def __init__(self, verbosity=log.LogLevel.Info, machine_only=False):
        self.stream = verbosity >= log.LogLevel.Trace
        self.verbosity = verbosity
        self.machine_only = machine_only
        self.mapping = {
            log.TestResult.type_id: self.handle_testresult,
            log.SuiteStatus.type_id: self.handle_suitestatus,
            log.TestStatus.type_id: self.handle_teststatus,
            log.TestStderr.type_id: self.handle_stderr,
            log.TestStdout.type_id: self.handle_stdout,
            log.TestMessage.type_id: self.handle_testmessage,
            log.LibraryMessage.type_id: self.handle_librarymessage,
        }

    def _display_outcome(self, name, outcome, reason=None):
        print(self.color.Bold
                 + SummaryHandler.colormap[outcome]
                 + name
                 + ' '
                 + state.Result.enums[outcome]
                 + SummaryHandler.reset)

        if reason is not None:
            log.test_log.info('')
            log.test_log.info('Reason:')
            log.test_log.info(reason)
            log.test_log.info(terminal.separator('-'))

    def handle_teststatus(self, record):
        if record['status'] == state.Status.Running:
            log.test_log.debug('Starting Test Case: %s' %\
                    record['metadata'].name)

    def handle_testresult(self, record):
        self._display_outcome(
            'Test: %s'  % record['metadata'].name,
            record['result'].value)

    def handle_suitestatus(self, record):
        if record['status'] == state.Status.Running:
              log.test_log.debug('Starting Test Suite: %s ' %\
                    record['metadata'].name)

    def handle_stderr(self, record):
        if self.stream:
            print(record.data['buffer'], file=sys.stderr, end='')

    def handle_stdout(self, record):
        if self.stream:
            print(record.data['buffer'], file=sys.stdout, end='')

    def handle_testmessage(self, record):
        if self.stream:
            print(self._colorize(record['message'], record['level']))

    def handle_librarymessage(self, record):
        if not self.machine_only or record.data.get('machine_readable', False):
            print(self._colorize(record['message'], record['level'],
                    record['bold']))

    def _colorize(self, message, level, bold=False):
        return '%s%s%s%s' % (
                self.color.Bold if bold else '',
                self.verbosity_mapping.get(level, ''),
                message,
                self.default)

    def handle(self, record):
        if record.data.get('level', self.verbosity) > self.verbosity:
            return
        self.mapping.get(record.type_id, lambda _:None)(record)

    def close(self):
        pass

class MultiprocessingHandlerWrapper(object):
    '''
    A handler class which forwards log records to subhandlers, enabling
    logging across multiprocessing python processes.

    The 'parent' side of the handler should execute either
    :func:`async_process` or :func:`process` to forward
    log records to subhandlers.
    '''
    def __init__(self, *subhandlers):
        # Create thread to spin handing recipt of messages
        # Create queue to push onto
        self.queue = multiprocessing.Queue()
        self.queue.cancel_join_thread()
        self._shutdown = threading.Event()

        # subhandlers should be accessed with the _handler_lock
        self._handler_lock = threading.Lock()
        self._subhandlers = subhandlers

    def add_handler(self, handler):
        self._handler_lock.acquire()
        self._subhandlers = (handler, ) + self._subhandlers
        self._handler_lock.release()

    def _with_handlers(self, callback):
        exception = None
        self._handler_lock.acquire()
        for handler in self._subhandlers:
            # Prevent deadlock when using this handler by delaying
            # exception raise until we get a chance to unlock.
            try:
                callback(handler)
            except Exception as e:
                exception = e
                break
        self._handler_lock.release()

        if exception is not None:
            raise exception

    def async_process(self):
        self.thread = threading.Thread(target=self.process)
        self.thread.daemon = True
        self.thread.start()

    def process(self):
        while not self._shutdown.is_set():
            try:
                item = self.queue.get(timeout=0.1)
                self._handle(item)
            except (KeyboardInterrupt, SystemExit):
                raise
            except EOFError:
                return
            except Empty:
                continue

    def _drain(self):
        while True:
            try:
                item = self.queue.get(block=False)
                self._handle(item)
            except (KeyboardInterrupt, SystemExit):
                raise
            except EOFError:
                return
            except Empty:
                return

    def _handle(self, record):
        self._with_handlers(lambda handler: handler.handle(record))

    def handle(self, record):
        self.queue.put(record)

    def _close(self):
        if hasattr(self, 'thread'):
            self.thread.join()
        _wrap(self._drain)
        self._with_handlers(lambda handler: _wrap(handler.close))

        # NOTE Python2 has an known bug which causes IOErrors to be raised
        # if this shutdown doesn't go cleanly on both ends.
        # This sleep adds some time for the sender threads on this process to
        # finish pickling the object and complete shutdown after the queue is
        # closed.
        time.sleep(.2)
        self.queue.close()
        time.sleep(.2)

    def close(self):
        if not self._shutdown.is_set():
            self._shutdown.set()
            self._close()


def _wrap(callback, *args, **kwargs):
    try:
        callback(*args, **kwargs)
    except:
        traceback.print_exc()