Source code for schrodinger.test.hypothesis.stateful

import sys
import traceback

import hypothesis
from hypothesis import stateful

from schrodinger.Qt import sip
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtCore import Qt
from schrodinger.infra import qt_message_handler


class RuleBasedStateMachineMetaClass(type(stateful.RuleBasedStateMachine)):
    """
    A metaclass that makes it easy to create simple rules for a
    RuleBasedStateMachine.

    'Simple' in this context means that the rule we want to create
        a) calls a method that takes no argument, and then
        b) doesn't check any results

    The only thing that we're testing with a rule like this is that calling the
    method never results in an unhandled exception. Ideally, we would check the
    results of calling methods, but even the lazy route is surprisingly useful
    in a state machine test. Advice: begin by adding as many of these simple
    rules as possible and then gradually replace them with richer rules and
    checks as the tests develop and experience indicates which checks are most
    valuable.

    Suppose that we are testing self.panel in our state machine and want to call
    the methods 'foo' on self.panel.bar and 'gar' on self.panel.mar.  We could
    write this::

        class PanelMachine(stateful.RuleBaseStateMachine):
            def __init__(self):
                super(PanelMachine, self).__init__()
                self.panel = MyPanel()

            @stateful.rule()
            def foo(self):
                self.panel.bar.foo()

            @stateful.rule()
            def gar(self):
                self.panel.mar.gar()

    But this quickly becomes tedious if we want to test a large number of such
    methods. Our metaclass will allow us to generate the desired rules for our
    state machine by adding strings to the SIMPLE_RULES tuple defined on the
    class::

        @six.add_metaclass(RuleBasedStateMachineMetaClass)
        class PanelMachine(stateful.RuleBaseStateMachine):
            SIMPLE_RULES = ('panel.bar.foo', 'panel.bar.gar')

        def __init__(self):
            super(PanelMachine, self).__init__()
            self.panel = MyPanel()
    """

    def __new__(cls, name, bases, dct):
        """
        Prior to regular class creation, iterates through the SIMPLE_RULES
        tuple defined on the state machine class and add stateful.rules for each
        of the items.
        """
        rules = dct.pop('SIMPLE_RULES', ())
        for path_name in rules:
            component_path = path_name.split(".")
            rule_name = "_".join(component_path)
            func = cls._makeFunc(component_path)
            # The name should be set for better test failure reporting
            func.__name__ = rule_name
            # decorate the function with the rule decorator
            dct[rule_name] = stateful.rule()(func)

        return super().__new__(cls, name, bases, dct)

    @classmethod
    def _makeFunc(cls, component_path):
        """
        Returns a test function that simply calls the method specified by
        component_path

        :param component_path: A list of strings picking out the component
        :type component_path: list of str

        :return: A callback that will call the method specified by
                 component_path
        :rtype: function
        """

        def func(self):
            """
            :return: A function that will simply call the specified method
            :rtype: function
            """
            target = self
            for name in component_path:
                target = getattr(target, name)
            if target is None:
                msg = "%s is not an attribute of your test class" % target
                raise ValueError(msg)
            target()

        return func


class QtErrorCatchingRuleBasedStateMachine(stateful.RuleBasedStateMachine):
    """
    A RuleBasedStateMachine that fails if any Qt warnings or errors are reported
    or if any exceptions are raised in a slot.  Without this class:

      - Qt warnings and errors are ignored.
      - Exceptions that are raised in a slot cause the test to fail with a
        "Exceptions caught outside of main thread:" message, but
        Hypothesis doesn't report the steps required to reproduce the exception.

    :note: The error checkers in this class are implemented in two parts.
        `_recordException` and `_recordQtWarningsAndErrors` record any problems
        that happen during a step, and the invariant `checkErrors` makes sure
        that no problems were recorded at the end of each step.  Any exception
        raised in one of the `_record` wouldn't be recognized by Hypothesis and
        would instead go directly to the top-level error handler, so the two
        step approach is required.

    :cvar FAIL_ON_QT_MSGS: What Qt message types should trigger a test failure.
        By default warnings, criticals, and fatals lead to a test failure while
        debug and info messages are ignored.  This can be modified in subclasses
        if desired.
    :vartype FAIL_ON_QT_MSGS: tuple or list or set

    :cvar QT_MSG_NAMES: A mapping from QtMsgType values to a textual
        description.
    :vartype QT_MSG_NAMES: dict(int, str)

    :cvar IGNORE_WARN_PREFIXES: Any Qt warning message that starts with a string
        listed here will be ignored. Warning messages filtered by
        qt_message_handler.filter_qt_msg will be ignored regardless of this var.
    :vartype IGNORE_WARN_PREFIXES: iterable(str)
    """

    FAIL_ON_QT_MSGS = (QtCore.QtWarningMsg, QtCore.QtCriticalMsg,
                       QtCore.QtFatalMsg)
    QT_MSG_NAMES = {
        QtCore.QtDebugMsg: "debug",
        QtCore.QtInfoMsg: "info",
        QtCore.QtWarningMsg: "warning",
        QtCore.QtCriticalMsg: "critical",
        QtCore.QtFatalMsg: "fatal"
    }
    IGNORE_WARN_PREFIXES = tuple()

    def __init__(self, trap_errors=True):
        """
        :param trap_errors: Whether to override the top-level exception handler
            and catch all Qt warnings and errors.  Setting this to False is
            *only* useful for interactive use of this object, i.e. instantiating
            one in a REPL.  Otherwise, all REPL tracebacks would be swallowed by
            the excepthook.
        :type trap_errors: bool
        """
        super().__init__()

        self._trap_errors = trap_errors
        self._exceptions = []
        self._qt_errors = []
        if trap_errors:
            # catch exceptions that get thrown in a slot
            self._old_excepthook = sys.excepthook
            sys.excepthook = self._recordException

            # catch Qt warnings and errors
            self._old_message_handler = QtCore.qInstallMessageHandler(
                self._recordQtWarningOrError)

    def teardown(self):
        """
        Reset the top-level exception handler and Qt's message handler once the
        test is done.
        """
        super().teardown()
        if self._trap_errors:
            sys.excepthook = self._old_excepthook
            QtCore.qInstallMessageHandler(self._old_message_handler)

    def _recordException(self, exc_type, value, tback):
        """
        Record any unhandled Python exceptions in `self._exceptions`.

        See sys.excepthook documentation
        (https://docs.python.org/3/library/sys.html#sys.excepthook) for argument
        documentation.
        """
        self._exceptions.append((exc_type, value, tback))

    def _recordQtWarningOrError(self, msg_type, context, msg):
        """
        Record any Qt warnings and errors in self._qt_errors.  Ignore warnings
        that start with anything in `IGNORE_WARN_PREFIXES`.

        See qInstallMessageHandler documentation
        (http://doc.qt.io/qt-5/qtglobal.html#qInstallMessageHandler) for
        argument documentation.
        """
        if msg_type in self.FAIL_ON_QT_MSGS:
            if qt_message_handler.filter_qt_msg(msg):
                return
            if msg_type == QtCore.QtWarningMsg:
                stripped = msg.lstrip()
                if any(
                        stripped.startswith(prefix)
                        for prefix in self.IGNORE_WARN_PREFIXES):
                    return
            self._qt_errors.append((msg_type, context, msg))

    @stateful.invariant()
    def checkErrors(self):
        """
        Fail if the current trial raised an unhandled exception or triggered a
        Qt warning or error.
        """
        self._checkErrors(self._exceptions, "Caught %i unhandled exceptions.",
                          self._formatTraceback)
        self._checkErrors(self._qt_errors, "%i Qt warnings or errors output.",
                          self._formatQtWarningOrError)

    def _checkErrors(self, errors, overflow_msg, error_formatter):
        """
        Raise an AssertionError if `errors` is not empty.

        :param errors: A list of errors
        :type errors: list[tuple]

        :param overflow_msg: If `errors` contains more than 10 errors, this
            message will be included in the AssertionError and only the first 10
            errors will be reported.
        :type overflow_msg: str

        :param error_formatter: A function that converts an element of `errors`
            into a text description.
        :type error_formatter: function
        """
        if not errors:
            return
        texts = []
        num_errors = len(self._qt_errors)
        if num_errors > 10:
            texts.append(overflow_msg % num_errors +
                         " Only reporting the first 10:")
            del errors[10:]
        texts.extend(error_formatter(*err) for err in errors)
        raise AssertionError("\n".join(texts))

    def _formatTraceback(self, exc_type, value, tback):
        """
        Return a text description of a traceback.

        See sys.excepthook documentation
        (https://docs.python.org/3/library/sys.html#sys.excepthook) for argument
        documentation.
        """
        msg = ''.join(traceback.format_tb(tback))
        msg += f'{exc_type.__name__}: {value}\n'
        return msg

    def _formatQtWarningOrError(self, msg_type, context, msg):
        """
        Return a text description of a Qt warning or error.

        See qInstallMessageHandler documentation
        (http://doc.qt.io/qt-5/qtglobal.html#qInstallMessageHandler) for
        argument documentation.
        """
        return "Qt {} output: {}".format(self.QT_MSG_NAMES[msg_type], msg)

    def _note(self, message: str):
        """
        Try to show the message as a note, which will only be shown as part of
        a failing recipe. If not in a test session (e.g. directly stepping
        through the stateful test), `hypothesis.note` raises an exception, so
        print the message instead.
        """
        try:
            hypothesis.note(message)
        except hypothesis.errors.InvalidArgument:
            print(message)


class PanelRuleBasedStateMachine(QtErrorCatchingRuleBasedStateMachine):
    """
    A RuleBasedStateMachine for use with a Qt-based panel.  Subclasses must
    define PANEL_CLASS, and the panel instance will be available as
    `self.panel`. Without this class:

    - QTimers may not fire when they are supposed to
    - Errors that occur during painting will not be caught
    - Errors in stateful decorator arguments may cause pytest to segfault

    :cvar PANEL_CLASS: The class of the panel to test.
    :vartype PANEL_CLASS: QtWidgets.QWidget
    """

    PANEL_CLASS = None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._app = QtWidgets.QApplication.instance()
        if self.PANEL_CLASS is None:
            raise RuntimeError("%s does not define PANEL_CLASS" %
                               self.__class__.__name__)
        self.panel = self._initPanel()
        # "show" the panel without actually showing it.  This allows the panel's
        # geometry to update as if it was shown on screen.
        self.panel.setAttribute(Qt.WA_DontShowOnScreen)
        self.panel.show()
        self._app.processEvents()

    def _initPanel(self):
        """
        Return an instance of the panel
        """
        panel = self.PANEL_CLASS()
        return panel

    def teardown(self):
        super().teardown()
        # The panel object must be completely destroyed before we start the next
        # Hypothesis run.  Otherwise, Hypothesis may not report recipes to
        # reproduce failures and may report flaky failures.  A deleteLater()
        # call followed by a processEvents() call isn't sufficient, since
        # processEvents() doesn't process DeferredDelete events (according to
        # the QCoreApplication::processEvents documentation).  Setting panel to
        # None and then calling gc.collect() is also insufficient, since that
        # doesn't guarantee destruction of the C++ objects.  To ensure that the
        # panel is really and truly dead, we use sip.delete().
        sip.delete(self.panel)
        self.panel = None

    def check_invariants(self):
        """
        Make sure we process events after each step so that any QTimers with
        timeouts of 0 will fire when they're supposed to.
        """
        for _ in range(5):
            self._app.processEvents()
        super().check_invariants()