pytest - How to log in celery while testing - Stack Overflow

admin管理员组

文章数量:1410725

I'm trying to test my celery tasks and can't get my log messages to show in the console when I test using the celery_worker fixture. I'm using celery v.5.4.

My tasks are defined in tasks.py

import logging
from celery import Celery

logger = logging.getLogger(__name__)

app = Celery("tasks", backend="rpc://", broker="pyamqp://")

@app.task()
def add(x, y):
    logger.debug("Adding %s and %s" % (x, y))
    return x + y

My test look something like this (task_test.py)

import logging
import tasks

logger = logging.getLogger(__name__)

def test_worker_basics(celery_worker):
    logger.debug("Testing basic task")
    assert tasks.add.delay(4, 4).get(timeout=10) == 8

def test_worker_basics_fail(celery_worker):
    logger.debug("Testing basic task fails")
    assert tasks.add.delay(4, 4).get(timeout=10) == 9

My conftest.py file looks like this

import pytest

pytest_plugins = ("celery.contrib.pytest", )

@pytest.fixture(scope="session")
def celery_config():
    return {
        "broker_url": "amqp://",
        "result_backend": "rpc://",
    }


@pytest.fixture(scope="session")
def celery_enable_logging():
    return True

I run the test with the following command

pytest -x task_test.py --log-cli-level DEBUG

I get DEBUG level log messages from celery itself, which looks like setup logs, but my own DEBUG level log messages are not there. If I include log messages with level ERROR they do show up. But INFO and WARNING also don't show.

DEBUG    celery.utils.functional:functional.py:335
def xmap(task, it):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def add(x, y):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def chain(*args, **kwargs):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def backend_cleanup():
    return 1

DEBUG    celery.utils.functional:functional.py:335
def xstarmap(task, it):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def chord(self, header, body, partial_args=0, interval=1, countdown=2, max_retries=3, eager=4, **kwargs):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def accumulate(self, *args, **kwargs):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def ping():
    return 1

DEBUG    celery.utils.functional:functional.py:335
def chunks(task, it, n):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def unlock_chord(self, group_id, callback, interval=0, max_retries=1, result=2, Result=3, GroupResult=4, result_from_tuple=5, **kwargs):
    return 1

DEBUG    celery.utils.functional:functional.py:335
def group(self, tasks, result, group_id, partial_args, add_to_parent=0):
    return 1

I tried using the celery_enable_logging fixture as described in the celery testing documentation, but I'm not sure if I'm doing it correctly and it doesn't solve the problem.

I also saw that there is a CELERYD_HIJACK_ROOT_LOGGER variable that can be set to False, which according to the celery configuration documentation should should allow me to configure my own logging handlers, but I'm not sure where I should put that.

本文标签： pytestHow to log in celery while testingStack Overflow