Use generator fixtures

celery/utils/pytest.py

@@ -93,16 +93,15 @@ def app(request):
     if is_not_contained:
         app.set_current()
 
-    def fin():
-        _state.set_default_app(prev_default_app)
-        _state._tls = prev_tls
-        _state._tls.current_app = prev_current_app
-        if app is not prev_current_app:
-            app.close()
-        _state._on_app_finalizers = prev_finalizers
-        _state._apps = prev_apps
-    request.addfinalizer(fin)
-    return app
+    yield app
+
+    _state.set_default_app(prev_default_app)
+    _state._tls = prev_tls
+    _state._tls.current_app = prev_current_app
+    if app is not prev_current_app:
+        app.close()
+    _state._on_app_finalizers = prev_finalizers
+    _state._apps = prev_apps
 
 
 @pytest.fixture()
@@ -111,15 +110,14 @@ def depends_on_current_app(app):
 
 
 @pytest.fixture(autouse=True)
-def reset_cache_backend_state(request, app):
-    def fin():
-        backend = app.__dict__.get('backend')
-        if backend is not None:
-            if isinstance(backend, CacheBackend):
-                if isinstance(backend.client, DummyClient):
-                    backend.client.cache.clear()
-                backend._cache.clear()
-    request.addfinalizer(fin)
+def reset_cache_backend_state(app):
+    yield
+    backend = app.__dict__.get('backend')
+    if backend is not None:
+        if isinstance(backend, CacheBackend):
+            if isinstance(backend.client, DummyClient):
+                backend.client.cache.clear()
+            backend._cache.clear()
 
 
 @decorator

t/unit/bin/test_worker.py

@@ -22,8 +22,9 @@ from celery.worker import state
 
 
 @pytest.fixture(autouse=True)
-def reset_worker_optimizations(request):
-    request.addfinalizer(trace.reset_worker_optimizations)
+def reset_worker_optimizations():
+    yield
+    trace.reset_worker_optimizations()
 
 
 class Worker(cd.Worker):

t/unit/conftest.py

@@ -36,7 +36,7 @@ CASE_LOG_HANDLER_EFFECT = 'Test {0} modified handlers for the root logger'
 
 
 @pytest.fixture(autouse=True, scope='session')
-def AAA_disable_multiprocessing(request):
+def AAA_disable_multiprocessing():
     # pytest-cov breaks if a multiprocessing.Process is started,
     # so disable them completely to make sure it doesn't happen.
     from case import patch
@@ -51,9 +51,9 @@ def AAA_disable_multiprocessing(request):
     ctxs = [patch(s) for s in stuff]
     [ctx.__enter__() for ctx in ctxs]
 
-    def fin():
-        [ctx.__exit__(*sys.exc_info()) for ctx in ctxs]
-    request.addfinalizer(fin)
+    yield
+
+    [ctx.__exit__(*sys.exc_info()) for ctx in ctxs]
 
 
 def alive_threads():
@@ -61,7 +61,7 @@ def alive_threads():
 
 
 @pytest.fixture(autouse=True)
-def task_join_will_not_block(request):
+def task_join_will_not_block():
     from celery import _state
     from celery import result
     prev_res_join_block = result.task_join_will_block
@@ -71,11 +71,11 @@ def task_join_will_not_block(request):
         _state.task_join_will_block = lambda: False
     _state._set_task_join_will_block(False)
 
-    def fin():
-        result.task_join_will_block = prev_res_join_block
-        _state.task_join_will_block = prev_state_join_block
-        _state._set_task_join_will_block(False)
-    request.addfinalizer(fin)
+    yield
+
+    result.task_join_will_block = prev_res_join_block
+    _state.task_join_will_block = prev_state_join_block
+    _state._set_task_join_will_block(False)
 
 
 @pytest.fixture(scope='session', autouse=True)
@@ -88,9 +88,8 @@ def record_threads_at_startup(request):
 
 @pytest.fixture(autouse=True)
 def threads_not_lingering(request):
-    def fin():
-        assert request.session._threads_at_startup == alive_threads()
-    request.addfinalizer(fin)
+    yield
+    assert request.session._threads_at_startup == alive_threads()
 
 
 @pytest.fixture(autouse=True)
@@ -110,10 +109,9 @@ def test_cases_shortcuts(request, app, patching):
         request.instance.CELERY_TEST_CONFIG = dict(CELERY_TEST_CONFIG)
         request.instance.add = add
         request.instance.patching = patching
-
-        def fin():
-            request.instance.app = None
-        request.addfinalizer(fin)
+    yield
+    if request.instance:
+        request.instance.app = None
 
 
 @pytest.fixture(autouse=True)
@@ -122,41 +120,45 @@ def zzzz_test_cases_calls_setup_teardown(request):
         # we set the .patching attribute for every test class.
         setup = getattr(request.instance, 'setup', None)
         # we also call .setup() and .teardown() after every test method.
-        teardown = getattr(request.instance, 'teardown', None)
         setup and setup()
-        teardown and request.addfinalizer(teardown)
+
+    yield
+
+    if request.instance:
+        teardown = getattr(request.instance, 'teardown', None)
+        teardown and teardown()
 
 
 @pytest.fixture(autouse=True)
-def sanity_no_shutdown_flags_set(request):
-    def fin():
-        # Make sure no test left the shutdown flags enabled.
-        from celery.worker import state as worker_state
-        # check for EX_OK
-        assert worker_state.should_stop is not False
-        assert worker_state.should_terminate is not False
-        # check for other true values
-        assert not worker_state.should_stop
-        assert not worker_state.should_terminate
-    request.addfinalizer(fin)
+def sanity_no_shutdown_flags_set():
+    yield
+
+    # Make sure no test left the shutdown flags enabled.
+    from celery.worker import state as worker_state
+    # check for EX_OK
+    assert worker_state.should_stop is not False
+    assert worker_state.should_terminate is not False
+    # check for other true values
+    assert not worker_state.should_stop
+    assert not worker_state.should_terminate
 
 
 @pytest.fixture(autouse=True)
 def sanity_stdouts(request):
-    def fin():
-        from celery.utils.log import LoggingProxy
-        assert sys.stdout
-        assert sys.stderr
-        assert sys.__stdout__
-        assert sys.__stderr__
-        this = request.node.name
-        if isinstance(sys.stdout, (LoggingProxy, Mock)) or \
-                isinstance(sys.__stdout__, (LoggingProxy, Mock)):
-            raise RuntimeError(CASE_LOG_REDIRECT_EFFECT.format(this, 'stdout'))
-        if isinstance(sys.stderr, (LoggingProxy, Mock)) or \
-                isinstance(sys.__stderr__, (LoggingProxy, Mock)):
-            raise RuntimeError(CASE_LOG_REDIRECT_EFFECT.format(this, 'stderr'))
-    request.addfinalizer(fin)
+    yield
+
+    from celery.utils.log import LoggingProxy
+    assert sys.stdout
+    assert sys.stderr
+    assert sys.__stdout__
+    assert sys.__stderr__
+    this = request.node.name
+    if isinstance(sys.stdout, (LoggingProxy, Mock)) or \
+            isinstance(sys.__stdout__, (LoggingProxy, Mock)):
+        raise RuntimeError(CASE_LOG_REDIRECT_EFFECT.format(this, 'stdout'))
+    if isinstance(sys.stderr, (LoggingProxy, Mock)) or \
+            isinstance(sys.__stderr__, (LoggingProxy, Mock)):
+        raise RuntimeError(CASE_LOG_REDIRECT_EFFECT.format(this, 'stderr'))
 
 
 @pytest.fixture(autouse=True)
@@ -165,14 +167,14 @@ def sanity_logging_side_effects(request):
     rootlevel = root.level
     roothandlers = root.handlers
 
-    def fin():
-        this = request.node.name
-        root_now = logging.getLogger()
-        if root_now.level != rootlevel:
-            raise RuntimeError(CASE_LOG_LEVEL_EFFECT.format(this))
-        if root_now.handlers != roothandlers:
-            raise RuntimeError(CASE_LOG_HANDLER_EFFECT.format(this))
-    request.addfinalizer(fin)
+    yield
+
+    this = request.node.name
+    root_now = logging.getLogger()
+    if root_now.level != rootlevel:
+        raise RuntimeError(CASE_LOG_LEVEL_EFFECT.format(this))
+    if root_now.handlers != roothandlers:
+        raise RuntimeError(CASE_LOG_HANDLER_EFFECT.format(this))
 
 
 def setup_session(scope='session'):

t/unit/worker/test_state.py

@@ -14,12 +14,11 @@ from celery.utils.collections import LimitedSet
 
 
 @pytest.fixture
-def reset_state(request):
-    def fin():
-        state.active_requests.clear()
-        state.revoked.clear()
-        state.total_count.clear()
-    request.addfinalizer(fin)
+def reset_state():
+    yield
+    state.active_requests.clear()
+    state.revoked.clear()
+    state.total_count.clear()
 
 
 class MockShelve(dict):