Work around Celery application state issues by nuking cached properties, reordering tests

Co-authored-by: cmessick <cmessick@users.noreply.github.com>
Co-authored-by: Shaon Ghosh <shaonghosh@gmail.com>
Co-authored-by: Geoffrey Mo <geoffreymo@users.noreply.github.com>
Co-authored-by: zadadam <zadadam@users.noreply.github.com>
Co-authored-by: santier14 <santier14@users.noreply.github.com>

gwcelery/tests/conftest.py

-from contextlib import contextmanager
 from unittest import mock
 
 from celery.contrib.testing.app import UnitLogging
+import celery.backends.cache
 import pytest
 from pytest_socket import disable_socket
 
@@ -12,14 +12,26 @@ from .process import starter  # noqa: F401
 def nuke_celery_backend():
     """Clear the cached Celery backend.
 
-    Some of our tests switch backend URLs. In order for the switch to take
-    effect, we need to make sure that the cached backed object has been reset.
+    The Celery application object caches a lot of instance members that are
+    affected by the application configuration. Since we are changing the
+    configuration between tests, we need to make sure that all of the cached
+    application state is reset.
+
+    FIXME: The pytest celery plugin does not seem like it is really designed
+    to use a pre-existing application object; it seems like it is designed to
+    create a test application.
 
     """
-    try:
-        del app._local.backend
-    except AttributeError:
-        pass
+    app._pool = None
+    for key in ['Worker', 'WorkController', 'Beat', 'Task', 'annotation',
+                'AsyncResult', 'ResultSet', 'GroupResult', 'oid', 'amqp',
+                'control', 'events', 'loader', 'log']:
+        try:
+            del app.__dict__[key]
+        except KeyError:
+            pass
+    celery.backends.cache._DUMMY_CLIENT_CACHE.clear()
+    app._local.__dict__.clear()
 
 
 @pytest.fixture
@@ -41,7 +53,7 @@ def update_celery_config():
     app.conf.update(tmp)
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture
 def noop_celery_config(reset_celery_backend, update_celery_config):
     """Ensure that the Celery app is disconnected from live services."""
     update_celery_config(dict(
@@ -97,5 +109,29 @@ def fake_legacy_gracedb_client(monkeypatch):
     monkeypatch.setattr('gwcelery.tasks.legacy_gracedb.client', mock_client)
 
 
-def pytest_runtest_setup():
+def pytest_configure(config):
+    config.addinivalue_line(
+        'markers', 'live_worker: run test using a live Celery worker')
+
+
+def pytest_collection_modifyitems(session, config, items):
+    # FIXME: We are reordering the tests so that the ones that use a live
+    # celery worker run last. Otherwise, they cause problems with other
+    # tests that use the 'caplog' fixture. We don't understand why this is.
+    # Remove this hack if and when we figure it out.
+    items[:] = sorted(
+        items,
+        key=lambda item: item.get_closest_marker('live_worker') is not None)
+
+
+def pytest_runtest_setup(item):
     disable_socket()
+
+
+@pytest.fixture(autouse=True)
+def maybe_celery_worker(request):
+    if request.node.get_closest_marker('live_worker') is None:
+        fixture = 'noop_celery_config'
+    else:
+        fixture = 'celery_worker'
+    request.getfixturevalue(fixture)

gwcelery/tests/test_tasks_condor.py

@@ -82,7 +82,8 @@ def mock_condor_submit(monkeypatch):
     monkeypatch.setattr('subprocess.run', mock_run)
 
 
-def test_check_output_error_on_submit(celery_worker, monkeypatch):
+@pytest.mark.live_worker
+def test_check_output_error_on_submit(monkeypatch):
     """Test capturing an error from condor_submit."""
     accounting_group = 'foo.bar'
     cmd = ('sleep', '1')
@@ -100,14 +101,16 @@ def test_check_output_error_on_submit(celery_worker, monkeypatch):
     assert exc_info.value.output == msg
 
 
-def test_check_output_aborted(celery_worker, mock_condor_submit_aborted):
+@pytest.mark.live_worker
+def test_check_output_aborted(mock_condor_submit_aborted):
     """Test a job that is aborted."""
     result = condor.check_output.delay(['sleep', '1'])
     with pytest.raises(condor.JobAborted):
         result.get()
 
 
-def test_check_output_fails(celery_worker, mock_condor_submit):
+@pytest.mark.live_worker
+def test_check_output_fails(mock_condor_submit):
     """Test a job that immediately fails."""
     result = condor.check_output.delay(['sleep', '--foo="bar bat"', '1'])
     with pytest.raises(condor.JobFailed) as exc_info:
@@ -115,14 +118,16 @@ def test_check_output_fails(celery_worker, mock_condor_submit):
     assert exc_info.value.returncode == 1
 
 
+@pytest.mark.live_worker
 @pytest.mark.skip("This test isn't working yet.")
-def test_check_output_running(celery_worker, mock_condor_submit_running):
+def test_check_output_running(mock_condor_submit_running):
     result = condor.check_output.delay(['sleep', '1'])
     with pytest.raises(condor.JobRunning):
         result.get(2)
 
 
-def test_check_output_succeeds(celery_worker, mock_condor_submit):
+@pytest.mark.live_worker
+def test_check_output_succeeds(mock_condor_submit):
     """Test a job that immediately succeeds."""
     result = condor.check_output.delay(['sleep', '1'])
     result.get()