title: Celery integration testing with pytest and monkeypatching
body: <p>I'm trying to run some simple integration tests with Python 3.9.5, Celery 5.2.6, pytest 7.1.0 and FastAPI on docker.</p>
  <p>Project structure:</p>
  <pre><code>  
    ┣ app
    ┣ ┣ api
    ┣ ┣ ┣ routes
    ┣ ┣ ┣ ┗ celery.py
    ┣ ┗ services
    ┣    ┗  operations.py
    ┣ ┣ celery
    ┣ ┃ ┣ tasks.py
    ┣ ┃ ┣ utils.py
    ┣ ┃ ┗ worker.py
    ┣ tests
       ┣ conftest.py
       ┗ test_celery.py
  </code></pre>
  <p>The celery container is run with: <code>command: celery -A app.celery.worker:celery_app worker --loglevel INFO --logfile=./celery.log --hostname=worker_1_dev@%h</code>, and works as expected.</p>
  <p>And the web server container with: <code>command: uvicorn app.api.server:app </code></p>
  <p>Both have bind mounts of the complete project directory.</p>
  <p>How should the celery fixtures be defined so that I can monkeypatch <code>operations.py</code> or the task itself?
  Right now when my tests run the actual worker container picks up the real tasks and runs them, so I guess the problem lies
  in the fixtures. Setting always eager does not change the behaviour, and isn't recommended either.
  The worker and app instance fixtures return objects along the lines of:</p>
  <pre><code>celery_worker: gen44@XXX
  celery_app: &lt;Celery celery.tests at XXX&gt;
  </code></pre>
  <p>which seems correct, they're not the actual worker container or the <code>app.celery.worker:celery_app</code> instance.
  I've tried messing with other fixtures in the docs, mocking approaches, etc. but I've spent way too much time on this and tests just
  don't seem to work. Like I said, the actual container executes tasks as expected.</p>
  <p><strong>app/api/routes/celery.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>class TaskStatus(CoreModel):
      task_id: str
      task_status: Optional[str]
      task_result: Optional[Any]
  
  @router.post(
    &quot;/operations/add-with-factor&quot;,
    name=&quot;celery:add-times-factor&quot;,
  )
  def add_times_factor(
    x: int = Query(..., gt=0),
    y: int = Query(..., gt=0),
  ):
    task = celery_app.send_task(
      &quot;add_times_factor&quot;,
      args=[x, y],
    )
    return {&quot;task_id&quot;: task.id}
  
  
  @router.get(
      &quot;/tasks/{task_id}/&quot;,
      name=&quot;celery:get-task-status&quot;,
      response_model=TaskStatus,
  )
  def get_status(task_id):
      task_result = AsyncResult(task_id)
      return TaskStatus(
          task_id=task_id,
          task_status=task_result.status,
          task_result=task_result.result,
      )
  </code></pre>
  <p><strong>tests/conftest.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>@pytest.fixture(scope=&quot;session&quot;)
  def celery_enable_logging():
      return True
  
  @pytest.fixture(scope=&quot;session&quot;)
  def celery_config():
      return {
          &quot;broker_url&quot;: &quot;memory://&quot;,
          &quot;result_backend&quot;: &quot;rpc://&quot;,
          &quot;task_always_eager&quot;: True, # to be removed
      }
  
  @pytest.fixture(scope=&quot;session&quot;)
  def celery_worker_parameters():
      return {
          &quot;perform_ping_check&quot;: False,
      }
  
  @pytest.fixture(scope=&quot;session&quot;)
  def celery_includes():
      return [&quot;app.celery.tasks&quot;]
  
      ...
  </code></pre>
  <p><strong>app/celery/worker.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>import os
  from celery import Celery
  
  celery_app = Celery(&quot;app.celery.worker&quot;)
  celery_app.conf.task_serializer = &quot;json&quot;
  celery_app.conf.result_serializer = &quot;json&quot;
  celery_app.conf.accept_content = [&quot;application/json&quot;, &quot;application/x-python-serialize&quot;]
  celery_app.conf.broker_url = os.environ.get(&quot;CELERY_BROKER_URL&quot;, &quot;redis://localhost:6379&quot;)
  celery_app.conf.result_backend = os.environ.get(&quot;CELERY_RESULT_BACKEND&quot;, &quot;redis://localhost:6379&quot;)
  celery_app.autodiscover_tasks([&quot;app.celery&quot;])
  </code></pre>
  <p><strong>app/celery/tasks.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>from app.services.operations import OperationService
  
  def async_to_sync(func):
      @functools.wraps(func)
      def wrapped(*args, **kwargs):
          return asyncio.run(func(*args, **kwargs))
  
      return wrapped
  
  @shared_task(name=&quot;add_times_factor&quot;)
  @async_to_sync
  async def add_times_factor_task(x, y):
      operation_service = OperationService()
      result = await operation_service.add_times_factor(x, y)
      return {&quot;result&quot;: result}
  
  </code></pre>
  <p><strong>app/services/operations.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>FACTOR = 2
  
  class OperationService:
      async def add_times_factor(self, x, y):
          await something
          return (x + y) * FACTOR
  </code></pre>
  <p><strong>tests/test_celery.py</strong></p>
  <pre class="lang-py prettyprint-override"><code>import pytest
  import app.services.operations as operations
  pytestmark = pytest.mark.asyncio
  
  class TestCelery:
      async def test_add_times_factor(self,
          celery_worker,
          celery_app: Celery,
          monkeypatch,
      ):
          monkeypatch.setattr(operations, &quot;FACTOR&quot;, 20) # ignored right now
  
          celery_app.set_current()
          res = await superuser_client.post(app.url_path_for(&quot;celery:add-times-factor&quot;), params={&quot;x&quot;: 1, &quot;y&quot;: 2})
  
          task_succeeded = False
          while not task_succeeded:
              task_status_res = await superuser_client.get(app.url_path_for(&quot;celery:get-task-status&quot;, task_id=task.id))
              assert task_status_res.status_code == HTTP_200_OK
              task_succeeded = task_status_res.json()[&quot;task_status&quot;] == &quot;SUCCESS&quot;
              await asyncio.sleep(0.5)
  
          assert json.loads(task_status_res.json()[&quot;task_result&quot;])[&quot;result&quot;] == 60 # fail 6 == 60
  
  
  </code></pre>