@cykooz ты научился правильно тестировать celery? А именно уйти от глобального celery_app
Ну то есть сейчас код такой:
celery = Celery(include=include)
@celery.task()
def empty_task():
logbook.info("Empty task is running")
return True
Получается celeryapp создается при импорте модуля, а соответственно и коннекшены к редису создаются. И в случае тестов получается два Celery объекта. Один который создаётся при импорте, а второй из фикстуры celery_app. Ну и самое противное, во время импорта модуля с тасками, у меня еще не мокан путь для редиса, который в unittestах используется. И получается что таски кладутся в очередь не unittestового редиса. А воркер в тестах слушает очередь в unittestовом redis'е.