diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2d1999a1278a9548fdd88da7fbdac02c07ba7839..be2e3f0f6e4ba72357d2fec32f715c2cb9a58b63 100644 --- a/CHANGELOG.rst +++ b/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 diff --git a/fsl/utils/idle.py b/fsl/utils/idle.py index afa6247aeb667fe41d75bbde57376fee31c500d3..d2c93aeb08e0928df4e0dcceb84adf32dc98550f 100644 --- a/fsl/utils/idle.py +++ b/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): diff --git a/tests/test_idle.py b/tests/test_idle.py index d97b19363da03edef7bfe472f0f8e34f831bd65b..fdee73ac7ab3ab93a02110587b6f397524df4767 100644 --- a/tests/test_idle.py +++ b/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():