More robust unit tests for async.idle(alwaysQueue=True). Also using a single

wx.App object for complex reasons

tests/test_async.py

@@ -9,29 +9,46 @@ import time
 import threading
 import random
 
+from six.moves import reload_module
+
 import pytest
+import mock
 
 import fsl.utils.async as async
 from fsl.utils.platform import platform as fslplatform
 
-
+# We use a single wx.App object because wx.GetApp()
+# will still return an old App objectd after its
+# mainloop has exited. and therefore async.idle
+# will potentially register on EVT_IDLE with the
+# wrong wx.App object.
+_wxapp = None
 def _run_with_wx(func, *args, **kwargs):
 
+    global _wxapp
+
     propagateRaise = kwargs.pop('propagateRaise', True)
     startingDelay  = kwargs.pop('startingDelay',  500)
     finishingDelay = kwargs.pop('finishingDelay', 500)
-    
-    import wx 
+    callAfterApp   = kwargs.pop('callAfterApp',   None)
+
+    import wx
 
     result = [None]
     raised = [None]
-    app    = wx.App()
+
+    if _wxapp is None:
+        _wxapp    = wx.App()
     frame  = wx.Frame(None)
-    
+
+    if callAfterApp is not None:
+        callAfterApp()
+
     def wrap():
 
         try:
-            result[0] = func(*args, **kwargs)
+            if func is not None:
+                result[0] = func(*args, **kwargs)
 
         except Exception as e:
             print(e)
@@ -40,14 +57,14 @@ def _run_with_wx(func, *args, **kwargs):
         finally:
             def finish():
                 frame.Destroy()
-                app.ExitMainLoop() 
+                _wxapp.ExitMainLoop()
             wx.CallLater(finishingDelay, finish)
     
     frame.Show()
 
     wx.CallLater(startingDelay, wrap)
 
-    app.MainLoop()
+    _wxapp.MainLoop()
     async.idleReset()
 
     if raised[0] and propagateRaise:
@@ -244,25 +261,98 @@ def test_idle_dropIfQueued():
     assert     task2called[0]
 
 
-def test_idle_alwaysQueue():
+def test_idle_alwaysQueue1():
 
+    # Test scheduling the task before
+    # a wx.App has been created.
     called = [False]
 
     def task():
         called[0] = True
 
+    # In this scenario, an additional call
+    # to idle (after the App has been created)
+    # is necessary, otherwise the originally
+    # queued task will not be called.
     def nop():
         pass
 
+    # The task should be run
+    # when the mainloop starts
     async.idle(task, alwaysQueue=True)
 
-    # We need to queue another task
-    # for the first task to be executed
+    # Second call to async.idle
     _run_with_wx(async.idle, nop)
 
     assert called[0]
 
 
+def test_idle_alwaysQueue2():
+
+    # Test scheduling the task
+    # after a wx.App has been craeted,
+    # but before MainLoop has started
+
+    called = [False]
+
+    def task():
+        called[0] = True
+
+    def queue():
+        async.idle(task, alwaysQueue=True)
+
+    _run_with_wx(None, callAfterApp=queue)
+
+    assert called[0]
+
+
+def test_idle_alwaysQueue3():
+
+    # Test scheduling the task
+    # after a wx.App has been craeted
+    # and the MainLoop has started.
+    # In this case, alwaysQueue should
+    # have no effect - the task should
+    # just be queued and executed as
+    # normal.
+
+    called = [False]
+
+    def task():
+        called[0] = True
+
+    _run_with_wx(async.idle, task, alwaysQueue=True)
+
+    assert called[0]
+
+
+def test_idle_alwaysQueue4():
+
+    # Test scheduling the task when
+    # wx is not present - the task
+    # should just be executed immediately
+    called = [False]
+
+    def task():
+        called[0] = True
+
+    import fsl.utils.platform
+    with mock.patch.dict('sys.modules', {'wx' : None}):
+
+        # async uses the platform module to
+        # determine whether a GUI is available,
+        # so we have to reload it
+        reload_module(fsl.utils.platform)
+        async.idle(task, alwaysQueue=True)
+
+        with pytest.raises(ImportError):
+            import wx
+
+    reload_module(fsl.utils.platform)
+
+    assert called[0]
+
+
 def test_idle_timeout():
 
     called = [False]