Source code for traits_futures.wx.event_loop_helper

# (C) Copyright 2018-2023 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.

import wx.lib.newevent

from traits_futures.i_event_loop_helper import IEventLoopHelper

# Note: we're not using the more obvious spelling
#   _SetattrEvent, _SetattrEventBinder = wx.lib.newevent.NewEvent()
# here because that confuses Sphinx's autodoc mocking.
# Ref: enthought/traits-futures#263.

#: New event type to be used for signalling attribute set operations.
_SetattrEventPair = wx.lib.newevent.NewEvent()
_SetattrEvent = _SetattrEventPair[0]
_SetattrEventBinder = _SetattrEventPair[1]

# XXX We should be using Pyface's own CallbackTimer instead of creating
# our own, but we were running into segfaults.
# xref: enthought/pyface#815, enthought/traits-futures#251
[docs]class TimeoutTimer(wx.Timer): """ Single-shot timer that executes a given callback on completion. Parameters ---------- timeout : float Timeout in seconds. callback Callable taking no arguments, to be executed when the timer times out. args : tuple, optional Tuple of positional arguments to pass to the callable. If not provided, no positional arguments are passed. kwargs : dict, optional Dictionary of keyword arguments to pass to the callable. If not provided, no keyword arguments are passed. """ def __init__(self, timeout, callback, args=(), kwargs=None): wx.Timer.__init__(self) self.timeout = timeout self.callback = callback self.args = args self.kwargs = {} if kwargs is None else kwargs
[docs] def start(self): """ Start the timer. """ timeout_in_ms = round(self.timeout * 1000) self.StartOnce(milliseconds=timeout_in_ms)
[docs] def stop(self): """ Stop the timer if it hasn't already expired. The callback will not be executed. """ self.Stop()
[docs] def Notify(self): """ Execute the callback when the timer completes. """ self.callback(*self.args, **self.kwargs)
[docs]class AppForTesting(wx.App): """ Subclass of wx.App used for testing. """
[docs] def OnInit(self): """ Override base class to ensure we have at least one window. """ # It's necessary to have at least one window to prevent the application # exiting immediately. self.frame = wx.Frame(None) self.SetTopWindow(self.frame) self.frame.Show(False) return True
[docs] def exit(self, exit_code): """ Exit the application main event loop with a given exit code. The event loop can be started and stopped several times for a single AppForTesting object. """ self.exit_code = exit_code self.ExitMainLoop()
[docs] def close(self): """ Clean up when the object is no longer needed. """ self.frame.Close() del self.frame
[docs]class AttributeSetter(wx.EvtHandler): """ Event handler that allows us to set object attributes from with a running event loop. """ def _on_setattr(self, event): setattr(event.obj,, event.value)
[docs]@IEventLoopHelper.register class EventLoopHelper: """ Support for running the wx event loop in unit tests. """
[docs] def init(self): """ Prepare the event loop for use. """ # Running tests requires that there's a visible application. self.wx_app = AppForTesting() attribute_setter = AttributeSetter() attribute_setter.Bind( _SetattrEventBinder, handler=attribute_setter._on_setattr ) self._attribute_setter = attribute_setter
[docs] def dispose(self): """ Dispose of any resources used by this object. """ attribute_setter = self._attribute_setter attribute_setter.Unbind( _SetattrEventBinder, handler=attribute_setter._on_setattr ) del self._attribute_setter self.wx_app.close() del self.wx_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. """ event = _SetattrEvent( obj=obj, name=name, value=value, ) wx.PostEvent(self._attribute_setter, event)
[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. """ wx_app = self.wx_app timeout_timer = TimeoutTimer(timeout, lambda: wx_app.exit(1)) def stop_if_condition(event): if condition(object): wx_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.start() try: wx_app.MainLoop() finally: timed_out = wx_app.exit_code timeout_timer.stop() 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) ) )