Source code for traits_futures.qt.event_loop_helper

# (C) Copyright 2018-2024 Enthought, Inc., Austin, TX
# All rights reserved.
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at
# Thanks for using Enthought open source!

Test support, providing the ability to run the event loop from tests.

from pyface.qt.QtCore import QObject, Qt, QTimer, Signal, Slot
from pyface.qt.QtGui import QApplication

from traits_futures.i_event_loop_helper import IEventLoopHelper

[docs] class AttributeSetter(QObject): """ Simple QObject that allows us to set object attributes from with a running event loop. """ @Slot(object, str, object) def _on_setattr(self, obj, name, value): """ Slot for setting an arbitrary attribute value on an object. """ setattr(obj, name, value) #: Signal used to trigger setattr operations. setattr = Signal(object, str, object)
[docs] @IEventLoopHelper.register class EventLoopHelper: """ Support for running the Qt event loop in unit tests. """
[docs] def init(self): """ Prepare the event loop for use. """ qt_app = QApplication.instance() if qt_app is None: qt_app = QApplication([]) self.qt_app = qt_app self._attribute_setter = AttributeSetter() self._attribute_setter.setattr.connect( self._attribute_setter._on_setattr, Qt.QueuedConnection )
[docs] def dispose(self): """ Dispose of any resources used by this object. """ self._attribute_setter.setattr.disconnect( self._attribute_setter._on_setattr ) del self._attribute_setter del self.qt_app
[docs] def setattr_soon(self, obj, name, value): """ Arrange for an attribute to be set once the event loop is running. In typical usage, *obj* will be a ``HasTraits`` instance and *name* will be the name of a trait on *obj*. This method is not thread-safe. It's designed to be called from the main thread. Parameters ---------- obj : object Object to set the given attribute on. name : str Name of the attribute to set; typically this is a traited attribute. value : object Value to set the attribute to. """ self._attribute_setter.setattr.emit(obj, name, value)
[docs] def run_until(self, object, trait, condition, timeout): """ Run event loop until the given condition holds true, or until timeout. The condition is re-evaluated, with the object as argument, every time the trait changes. Parameters ---------- object : traits.has_traits.HasTraits Object whose trait we monitor. trait : str Name of the trait to monitor for changes. condition Single-argument callable, returning a boolean. This will be called with *object* as the only input. timeout : float Number of seconds to allow before timing out with an exception. Raises ------ RuntimeError If timeout is reached, regardless of whether the condition is true or not at that point. """ qt_app = self.qt_app timeout_in_ms = round(1000.0 * timeout) timeout_timer = QTimer() timeout_timer.setSingleShot(True) timeout_timer.setInterval(timeout_in_ms) def stop_on_timeout(): qt_app.exit(1) def stop_if_condition(event): if condition(object): qt_app.exit(0) object.observe(stop_if_condition, trait) try: # The condition may have become True before we # started listening to changes. So start with a check. if condition(object): timed_out = 0 else: timeout_timer.timeout.connect(stop_on_timeout) timeout_timer.start() try: # Qt wrapper support is a bit of a mess here. PyQt6 has # only "exec". PyQt5 and PySide6 support both "exec" and # "exec_", with "exec" the preferred spelling. PyQt4 and # PySide2 support only "exec_". try: exec_method = qt_app.exec except AttributeError: exec_method = qt_app.exec_ timed_out = exec_method() finally: timeout_timer.stop() timeout_timer.timeout.disconnect(stop_on_timeout) finally: object.observe(stop_if_condition, trait, remove=True) if timed_out: raise RuntimeError( "run_until timed out after {} seconds. " "At timeout, condition was {}.".format( timeout, condition(object) ) )