Coverage for django_napse/core/celery_app.py: 75%

1import os 

2from unittest import mock 

3 

4import celery 

5from celery.app import trace 

6from celery.signals import setup_logging 

7from django.conf import settings 

8 

9# set the default Django settings module for the 'celery' program. 

10os.environ.setdefault("DJANGO_SETTINGS_MODULE", "tests.test_app.settings") 

11celery_app = celery.Celery("django_napse") 

12 

13 

14# Using a string here means the worker doesn't have to serialize 

15# the configuration object to child processes. 

16# - namespace='CELERY' means all celery-related configuration keys 

17# should have a `CELERY_` prefix. 

18celery_app.config_from_object(settings, namespace="CELERY") 

19 

20# Gestion des logs ------------------------------------------------------------------ 

21# Remove the trace from celery logs 

22old_info = trace.info 

23trace.info = mock.Mock() 

24 

25 

26# Configure celery logging 

27@setup_logging.connect 

28def config_loggers(*args: list, **kwargs: dict) -> None: # noqa: ARG001 

29 """Take settings.LOGGING as config for celery loggers (especially for `formatters`).""" 

30 from logging.config import dictConfig 

31 

32 from django.conf import settings 

33 

34 dictConfig(settings.LOGGING) 

35 

36 

37# Remove the strategy from celery logs 

38def strategy_log_free(*args: list, **kwargs: dict) -> callable: 

39 """Remove the strategy from celery logs.""" 

40 kwargs["info"] = mock.Mock() 

41 return celery.worker.strategy.default(*args[1:], **kwargs) 

42 

43 

44# Load task modules from all registered Django app configs. 

45celery_app.autodiscover_tasks()