Merge branch 'rf/idle_sync_ctxmgrs' into 'master'

Rf/idle sync ctxmgrs

See merge request fsl/fslpy!174

CHANGELOG.rst

@@ -13,6 +13,8 @@ Added
 * New ``until`` option to the :func:`.idle.block` function.
 * New :meth:`.Idle.neverQueue` setting, which can be used to force all
   tasks passed to :func:`.idle.idle` to be executed synchronously.
+* New :meth:`.IdleLoop.synchronous` context manager, to temporarily change the
+  value of :meth:`.IdleLoop.neverQueue`.
 
 
 Changed

fsl/utils/idle.py

@@ -79,6 +79,7 @@ import atexit
 import logging
 import functools
 import threading
+from   contextlib  import contextmanager
 from   collections import abc
 
 try:                import queue
@@ -225,15 +226,31 @@ class IdleLoop(object):
     @property
     def neverQueue(self):
         """If ``True``, tasks passed to :meth:`idle` will never be queued, and
-        instead will always be executed directly/synchonously.
+        instead will always be executed directly/synchonously. See also the
+        :meth:`synchronous` context manager.
         """
         return self.__neverQueue
 
 
     @neverQueue.setter
-    def neverQueue(self, allow):
+    def neverQueue(self, val):
         """Update the ``neverQueue`` flag. """
-        self.__neverQueue = allow
+        self.__neverQueue = val
+
+
+    @contextmanager
+    def synchronous(self):
+        """Context manager which can be used to tenporarily set :meth:`neverQueue` to
+        ``True``, restoring its previous value afterwards.
+        """
+
+        oldval = self.__neverQueue
+        self.__neverQueue = True
+
+        try:
+            yield
+        finally:
+            self.__neverQueue = oldval
 
 
     def reset(self):

tests/test_idle.py

@@ -427,6 +427,65 @@ def test_idle_alwaysQueue4():
     assert called[0]
 
 
+@pytest.mark.wxtest
+def test_neverQueue(): _run_with_wx(_test_neverQueue)
+def _test_neverQueue():
+    called = [False]
+    def task():
+        called[0] = True
+
+    oldval = idle.idleLoop.neverQueue
+
+    try:
+        idle.idleLoop.neverQueue = True
+
+        idle.idle(task)
+        assert called[0]
+
+        idle.idleLoop.neverQueue = False
+        called[0] = False
+        idle.idle(task)
+
+        assert not called[0]
+        _wait_for_idle_loop_to_clear()
+        assert called[0]
+
+    finally:
+        idle.idleLoop.neverQueue = oldval
+
+
+@pytest.mark.wxtest
+def test_synchronous(): _run_with_wx(_test_synchronous)
+def _test_synchronous():
+
+    called = [False]
+    def task():
+        called[0] = True
+
+    def test_async():
+        called[0] = False
+        idle.idle(task)
+        assert not called[0]
+        _wait_for_idle_loop_to_clear()
+        assert called[0]
+
+    oldval = idle.idleLoop.neverQueue
+
+    try:
+        idle.idleLoop.neverQueue = False
+        test_async()
+
+        with idle.idleLoop.synchronous():
+            called[0] = False
+            idle.idle(task)
+            assert called[0]
+
+        test_async()
+
+    finally:
+        idle.idleLoop.neverQueue = oldval
+
+
 @pytest.mark.wxtest
 def test_idle_timeout():