Add GuiTestAssistant.exercise_event_loop (#377)

* Add GuiTestAssistant.exercise_event_loop

* Remove unused timeout

* Update traits_futures/testing/gui_test_assistant.py

Co-authored-by: Poruri Sai Rahul <rporuri@enthought.com>

Co-authored-by: Poruri Sai Rahul <rporuri@enthought.com>

traits_futures/testing/gui_test_assistant.py

@@ -13,6 +13,8 @@
 """
 
 
+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
@@ -21,6 +23,17 @@
 SAFETY_TIMEOUT = 5.0
 
 
+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)
+
+
 class GuiTestAssistant:
     """
     Convenience mixin class for tests that need the event loop.
@@ -73,3 +86,21 @@ def run_until(self, object, trait, condition, timeout=SAFETY_TIMEOUT):
             true or not at that point.
         """
         self._event_loop_helper.run_until(object, trait, condition, timeout)
+
+    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)

traits_futures/tests/i_pingee_tests.py

@@ -17,7 +17,7 @@
 import threading
 import weakref
 
-from traits.api import Bool, Event, HasStrictTraits, Int
+from traits.api import Event, HasStrictTraits, Int
 
 #: Safety timeout, in seconds, for blocking operations, to prevent
 #: the test suite from blocking indefinitely if something goes wrong.
@@ -180,16 +180,7 @@ def send_delayed_ping(pingee, ready):
         # There shouldn't be any ping-related activity queued on the event
         # loop at this point. We exercise the event loop, in the hope
         # of flushing out any such activity.
-
-        class Sentinel(HasStrictTraits):
-            #: Simple boolean flag.
-            flag = Bool(False)
-
-        sentinel = Sentinel()
-
-        self._event_loop_helper.setattr_soon(sentinel, "flag", True)
-        self.run_until(sentinel, "flag", lambda sentinel: sentinel.flag)
-
+        self.exercise_event_loop()
         self.assertEqual(listener.ping_count, 0)
 
     def test_disconnect_removes_callback_reference(self):

traits_futures/tests/test_gui_test_assistant.py

@@ -12,7 +12,7 @@
 Tests for the GuiTestAssistant.
 """
 import time
-import unittest
+import unittest.mock
 
 from traits.api import Event, HasStrictTraits
 
@@ -127,3 +127,17 @@ def test_run_until_success(self):
             condition=lambda executor: executor.stopped,
         )
         self.assertTrue(executor.stopped)
+
+    def test_exercise_event_loop(self):
+        # There's little that we can usefully test here: exercising the event
+        # loop is *likely* to flush out pending events, but there are no
+        # *guaranteed* observable changes from exercising the event loop. So we
+        # merely call the method to check that it exists, and check that
+        # the event loop helper's run_until was called as a side-effect.
+        with unittest.mock.patch.object(
+            self._event_loop_helper,
+            "run_until",
+            wraps=self._event_loop_helper.run_until,
+        ) as mock_run_until:
+            self.exercise_event_loop()
+        mock_run_until.assert_called()

traits_futures/tests/traits_executor_tests.py

@@ -463,24 +463,6 @@ def target(executor, msg_queue):
 
     # Helper methods and assertions ###########################################
 
-    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.
-        """
-
-        class Sentinel(HasStrictTraits):
-            #: Simple boolean flag.
-            flag = Bool(False)
-
-        sentinel = Sentinel()
-
-        self._event_loop_helper.setattr_soon(sentinel, "flag", True)
-        self.run_until(sentinel, "flag", lambda sentinel: sentinel.flag)
-
     def wait_until_stopped(self, executor):
         """
         Wait for the executor to reach STOPPED state.