18👍
OP here, I’ve figured it out and wrote an article – https://medium.com/@scythargon/how-to-use-celery-pytest-fixtures-for-celery-intergration-testing-6d61c91775d9
Main key:
@pytest.mark.usefixtures('celery_session_app')
@pytest.mark.usefixtures('celery_session_worker')
class MyTest():
def test(self):
assert mul.delay(4, 4).get(timeout=10) == 16
4👍
For how many developers use these tools, I’m surprised at how lacking the docs are on the topic. I struggled with this for about a half day and then found @scythargon‘s discussion. I solved it slightly differently, so I’m throwing my answer in the the mix for posterity (very close to the OP’s method):
tasks.py
from celery import shared_task
@shared_task
def add(x, y):
return x + y
@shared_task()
def multiply(x, y):
return x * y
conftest.py
import pytest
pytest_plugins = ('celery.contrib.pytest', )
@pytest.fixture(scope='session')
def celery_config():
return {
'broker_url': 'redis://localhost:8001',
'result_backend': 'redis://localhost:8001'
}
tests.py
from api.app.tasks import add, multiply
def test_celery_worker_initializes(celery_app, celery_worker):
assert True
def test_celery_tasks(celery_app, celery_worker):
assert add.delay(4, 4).get(timeout=5) == 8
assert multiply.delay(4, 4).get(timeout=5) == 16
As an added bonus, my redis
broker and backend are running in Docker (as part of a swarm):
docker-compose.yml
version: "3.9"
services:
. . .
redis:
image: redis:alpine
networks:
swarm-net:
aliases:
- redis
ports:
- "8001:6379"
. . .
- How to specify uniqueness for a tuple of field in a Django model
- How should multiple Django apps communicate with each other?
- (gcloud.app.deploy) Error Response: [7] Access Not Configured. Cloud Build has not been used in project
- How to test coverage properly with Django + Nose
- Set db per model in django
Source:stackexchange.com