parallel - test post django

Haga que la base de datos de casos de prueba de Django sea visible para Apio

Question

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 retrotraigan cuando finaliza cada prueba. Intento crear una prueba de integración con Apio, pero no puedo encontrar la manera de conectar Apio 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 permanecen 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:

1. ¿Cómo se hace para que Aplery pueda ver los objetos que el caso de prueba Django?

2. ¿Cómo me aseguro de que todos los objetos guardados por Celery se retrotraigan automáticamente una vez que se complete la prueba?

Estoy dispuesto a limpiar manualmente los objetos si esto no es posible automáticamente, pero una eliminación de objetos en tearDown incluso en APISimpleTestCase parece que se ha revertido.

Answer 1

Para sus pruebas de unidad, recomendaría omitir la dependencia del apio, los dos siguientes enlaces le proporcionarán las informaciones necesarias para comenzar sus pruebas de unidad:

• http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html
• http://docs.celeryproject.org/en/latest/userguide/testing.html

Si realmente quieres probar las llamadas a la función del apio, incluida una cola, establecería un dockercompose de forma confiable con el servidor, el trabajador, la combinación de cola y extenderé el CeleryTestRunner personalizado desde los documentos de django-apio. Pero no vería un beneficio de esto porque el sistema de prueba es muy alejado de la producción para ser representativo.

Answer 2

Esto es posible al iniciar un trabajador de Apio dentro del caso de prueba de Django.

Fondo

La base de datos en memoria de Django es sqlite3. Como dice en la página de descripción de las bases de datos en memoria Sqlite , "[ Todas las conexiones de bases de datos que comparten la base de datos en memoria deben estar en el mismo proceso". Esto significa que, mientras Django use una base de datos de prueba en memoria y Celery se inicie en un proceso separado, es fundamentalmente imposible que Celery y Django compartan una base de datos de prueba.

Sin embargo, con celery.contrib.testing.worker.start_worker , es posible iniciar a un trabajador de Apio en un hilo separado dentro del mismo proceso. Este trabajador puede acceder a la base de datos en memoria.

Esto supone que Celery ya está configurado de la manera habitual con el proyecto Django.

Solución

Debido a que Django-Aplery implica una comunicación entre hilos, solo los casos de prueba que no se ejecutan en transacciones aisladas funcionarán. El caso de prueba debe heredar directamente de SimpleTestCase o su equivalente APISimpleTestCase equivalente y establecer el atributo de clase allow_database_queries en True .

La clave es iniciar a un trabajador de Apio en el método setUpClass del TestCase y cerrarlo en el método tearDownClass . La función clave es celery.contrib.testing.worker.start_worker(app) , que requiere una instancia de la aplicación actual de Apio, supuestamente obtenida de mysite.celery.app y devuelve un Python ContextManager , que tiene los métodos __enter__ y __exit__ , que deben ser llamado en setUpClass y tearDownClass , respectivamente. Probablemente haya una manera de evitar ingresar manualmente y ContextManager con un decorador o algo así, pero no pude resolverlo. Aquí hay un ejemplo de archivo de tests.py :

from celery.contrib.testing.worker import start_worker
from django.test import SimpleTestCase

from mysite.celery import app

class BatchSimulationTestCase(SimpleTestCase):
    allow_database_queries = True

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # Start up celery worker
        cls.celery_worker = start_worker(app)
        cls.celery_worker.__enter__()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()

        # Close worker
        cls.celery_worker.__exit__(None, None, None)

    def test_my_function(self):
        # my_task.delay() or something

Por alguna razón, el trabajador de prueba intenta usar una tarea llamada ''celery.ping'' , probablemente para proporcionar mejores mensajes de error en caso de falla del trabajador. Incluso establecer perform_ping_check en False como un argumento de palabra clave ot start_worker aún prueba su existencia. La tarea que está buscando es celery.contrib.testing.tasks.ping . Sin embargo, esta tarea no está instalada de manera predeterminada. Debería ser posible proporcionar esta tarea agregando celery.contrib.testing a INSTALLED_APPS en settings.py . Sin embargo, esto solo lo hace visible para el trabajador; no el código que genera el trabajador. El código que genera el trabajador hace una assert ''celery.ping'' in app.tasks , que falla. Al comentar esto, todo funciona, pero modificar una biblioteca instalada no es una buena solución. Probablemente estoy haciendo algo mal, pero la solución que establecí es copiar la función simple en algún lugar que pueda ser recogida por app.autodiscover_tasks() , como celery.py :

@app.task(name=''celery.ping'')
def ping():
    # type: () -> str
    """Simple task that just returns ''pong''."""
    return ''pong''

Ahora, cuando se ejecutan las pruebas, no hay necesidad de comenzar un proceso de Apio por separado. Un trabajador de Apio se iniciará en el proceso de prueba de Django como un hilo separado. Este trabajador puede ver las bases de datos en memoria, incluida la base de datos de prueba predeterminada en memoria. Para controlar el número de trabajadores, hay opciones disponibles en start_worker , pero parece que el valor predeterminado es un solo trabajador.