django - software - documentation celery




Rendre la base de données de cas de test Django visible à Celery (2)

Lorsqu'un cas de test Django s'exécute, il crée une base de données de test isolée afin que les écritures de base de données soient annulées à la fin de chaque test. J'essaie de créer un test d'intégration avec Celery, mais je n'arrive pas à comprendre comment connecter Celery à cette base de données de test éphémère. Dans la configuration naïve, les objets enregistrés dans Django sont invisibles pour Celery et les objets enregistrés dans Céleri persistent indéfiniment.

Voici un exemple de cas de test:

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

J'ai deux questions:

  1. Comment faire pour que Celery puisse voir les objets que le cas de test de Django?

  2. Comment puis-je m'assurer que tous les objets sauvegardés par Celery sont automatiquement annulés une fois le test terminé?

Je suis prêt à nettoyer manuellement les objets si cela est automatique n'est pas possible, mais une suppression d'objets dans tearDown même dans APISimpleTestCase semble être APISimpleTestCase .


Pour vos tests unititaires, je recommanderais de sauter la dépendance au céleri, les deux liens suivants vous fourniront les informations nécessaires pour démarrer votre unittests:

Si vous voulez vraiment tester les appels de fonction de céleri, y compris une file d'attente, j'ai probablement mis en place un dockercompose avec le serveur, l'ouvrier, la file d'attente et étendez le CeleryTestRunner personnalisé à partir des docs django-celery. Mais je n'en verrais aucun avantage, car le système de test est trop éloigné de la production pour être représentatif.


Ceci est possible en démarrant un worker Celery dans le cas de test de Django.

Contexte

La base de données en mémoire de Django est sqlite3. Comme il est indiqué sur la page de description des bases de données en mémoire Sqlite , "[ T] outes les connexions de base de données partageant la base de données en mémoire doivent être dans le même processus". Cela signifie que, tant que Django utilise une base de données de test en mémoire et que Celery est lancé dans un processus séparé, il est fondamentalement impossible d'avoir Celery et Django pour partager une base de données de test.

Cependant, avec celery.contrib.testing.worker.start_worker , il est possible de démarrer un worker Celery dans un thread séparé au sein du même processus. Ce travailleur peut accéder à la base de données en mémoire.

Cela suppose que Celery est déjà configuré de la manière habituelle avec le projet Django.

Solution

Étant donné que Django-Celery implique une communication inter-thread, seuls les cas de test qui ne s'exécutent pas dans des transactions isolées fonctionneront. Le APISimpleTestCase test doit hériter directement de SimpleTestCase ou de son équivalent APISimpleTestCase et définir l'attribut de classe allow_database_queries sur True .

La clé consiste à démarrer un worker Celery dans la méthode setUpClass de TestCase et à le fermer dans la méthode tearDownClass . La fonction clé est celery.contrib.testing.worker.start_worker(app) , qui nécessite une instance de l'application Celery actuelle, probablement obtenue à partir de mysite.celery.app et retourne un Python ContextManager , qui a des méthodes __enter__ et __exit__ , qui doivent être appelé dans setUpClass et tearDownClass , respectivement. Il y a probablement un moyen d'éviter d'entrer et d'entrer manuellement le ContextManager avec un décorateur ou quelque chose, mais je ne pouvais pas le comprendre. Voici un exemple de fichier 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

Pour quelque raison que ce soit, le 'celery.ping' tente d'utiliser une tâche appelée 'celery.ping' , probablement pour fournir de meilleurs messages d'erreur en cas d'échec du worker. Même la définition de perform_ping_check sur False tant qu'argument mot-clé de start_worker teste encore son existence. La tâche recherchée est celery.contrib.testing.tasks.ping . Toutefois, cette tâche n'est pas installée par défaut. Il devrait être possible de fournir cette tâche en ajoutant celery.contrib.testing à INSTALLED_APPS dans settings.py . Cependant, cela ne fait que le rendre visible pour le travailleur. pas le code qui génère le travailleur. Le code qui génère le worker fait une assert 'celery.ping' in app.tasks , ce qui échoue. Commenter cela fait tout fonctionner, mais modifier une bibliothèque installée n'est pas une bonne solution. Je fais probablement quelque chose de mal, mais la solution de contournement sur laquelle je me suis app.autodiscover_tasks() est de copier la fonction simple quelque part, elle peut être récupérée par app.autodiscover_tasks() , comme celery.py :

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

Maintenant, lorsque les tests sont exécutés, il n'est pas nécessaire de démarrer un processus de céleri séparé. Un ouvrier céleri sera démarré dans le processus de test Django en tant que thread séparé. Ce travailleur peut voir toutes les bases de données en mémoire, y compris la base de données de test en mémoire par défaut. Pour contrôler le nombre de travailleurs, il existe des options disponibles dans start_worker , mais il semble que la valeur par défaut soit un seul travailleur.