Source code for traits_futures.testing.test_assistant

# (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 within tests.

import asyncio

from traits.api import Bool, HasStrictTraits

from traits_futures.asyncio.event_loop import AsyncioEventLoop

#: Maximum timeout for blocking calls, in seconds. A successful test should
#: never hit this timeout - it's there to prevent a failing test from hanging
#: forever and blocking the rest of the test suite.

class _HasBool(HasStrictTraits):
    Simple HasTraits class with a single mutable trait.

    Used in tests that need something mutable and observable.

    #: Simple boolean flag.
    flag = Bool(False)

[docs] class TestAssistant: """ Convenience mixin class for tests that need the event loop. This class is designed to be used as a mixin alongside unittest.TestCase for tests that need to run the event loop as part of the test. Most of the logic is devolved to a toolkit-specific EventLoopHelper class. """
[docs] def event_loop_factory(self): """ Factory for the event loop. Returns ------- event_loop: IEventLoop """ asyncio_event_loop = asyncio.new_event_loop() self.addCleanup(asyncio_event_loop.close) return AsyncioEventLoop(event_loop=asyncio_event_loop)
[docs] def setUp(self): self._event_loop = self.event_loop_factory() self._event_loop_helper = self._event_loop.helper() self._event_loop_helper.init()
[docs] def tearDown(self): self._event_loop_helper.dispose() del self._event_loop_helper del self._event_loop
[docs] def run_until(self, object, trait, condition, timeout=SAFETY_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, optional Number of seconds to allow before timing out with an exception. The (somewhat arbitrary) default is 5 seconds. Raises ------ RuntimeError If timeout is reached, regardless of whether the condition is true or not at that point. """ self._event_loop_helper.run_until(object, trait, condition, timeout)
[docs] def exercise_event_loop(self): """ Exercise the event loop. Places a new task on the event loop and runs the event loop until that task is complete. The goal is to flush out any other tasks that might already be in event loop tasks queue. Note that there's no guarantee that this will execute other pending event loop tasks. So this method is useful for tests of the form "check that nothing bad happens as a result of other pending event loop tasks", but it's not safe to use it for tests that *require* pending event loop tasks to be processed. """ sentinel = _HasBool() self._event_loop_helper.setattr_soon(sentinel, "flag", True) self.run_until(sentinel, "flag", lambda sentinel: sentinel.flag)