Cuando se ejecuta un caso de prueba de Django, crea una base de datos de prueba aislada para que las escrituras de la base de datos se reviertan cuando se completa cada prueba. Estoy tratando de crear una prueba de integración con Celery, pero no puedo entender cómo conectar Celery a esta base de datos de prueba efímera. En la configuración ingenua, los objetos guardados en Django son invisibles para Apio y los objetos guardados en Apio persisten indefinidamente.

Aquí hay un ejemplo de caso de prueba:

import json
from rest_framework.test import APITestCase
from myapp.models import MyModel
from myapp.util import get_result_from_response

class MyTestCase(APITestCase):
    @classmethod
    def setUpTestData(cls):
        # This object is not visible to Celery
        MyModel(id='test_object').save()

    def test_celery_integration(self):
        # This view spawns a Celery task
        # Task should see MyModel.objects.get(id='test_object'), but can't
        http_response = self.client.post('/', 'test_data', format='json')

        result = get_result_from_response(http_response)
        result.get()  # Wait for task to finish before ending test case
        # Objects saved by Celery task should be deleted, but persist

Tengo dos preguntas:

¿Cómo lograr que Celery pueda ver los objetos que el caso de prueba de Django?

¿Cómo me aseguro de que todos los objetos guardados por Celery se deshagan automáticamente una vez que se completa la prueba?

Estoy dispuesto a limpiar manualmente los objetos si no es posible hacerlo automáticamente, pero se eliminarán los objetos entearDown incluso enAPISimpleTestCase parece ser revertido.