Python-сообщество

Headmaster · Март 16, 2022 20:30:07

Есть условная celery таска

 class CreateNotifications(Task):
    name = 'create-notifications'
 
    def run(self,  *args, **kwargs):
        # код таски

Хочу запустить её в тестах (не замокать, а именно запустить) в синхронном режиме, как обычную функцию, чтобы не требовался redis или другой брокер.

Как это сделать?

Думал использовать вот такую штуку https://docs.celeryq.dev/en/v5.2.3/userguide/testing.html#celery-worker-embed-live-worker

 import pytest
 
@pytest.mark.django_db
class TestClass:
    def test_celery_notification_create(self, celery_worker, user):
        # some test logic

но получаю ошибку psycopg2.InterfaceError: connection already closed

В чём может быть проблема?

VadimK · Март 24, 2022 01:06:24

celery задача запускается через .delay() или .async()
https://docs.celeryq.dev/en/stable/userguide/tasks.html

 @app.task
def add(x, y):
    return x + y

Вот запуск как задачи

 add.delay(8, 8)

А вот запуск как обычной функции.

 add(8, 8)

Отредактировано VadimK (Март 24, 2022 01:06:41)

Headmaster · Март 24, 2022 09:39:16

VadimK
celery задача запускается через .delay() или .async() https://docs.celeryq.dev/en/stable/userguide/tasks.html

Так это если таска написана как функция, тут я и сам знаю как делать. А если таска это класс? (пример есть у меня в певром посте)

VadimK · Март 24, 2022 18:57:22

А разве класс не запускается точно таким же образом ? Никогда на классах не делал. Сейчас поискал в гугле, вижу что запуск так же идет как и у функций. Только указывается сам класс а не run()
https://jsatt.com/blog/class-based-celery-tasks/

Headmaster · Март 24, 2022 19:23:14

VadimK
https://jsatt.com/blog/class-based-celery-tasks/

Не запускается. Сам когда гуглил, находил эту статью и ничего

Я глянул исходники celery.Task - запуска не через брокер не нашёл. Либо плохо искал, либо на самом деле нету.

Касательно pytest - там есть несколько fixtures для работы с celery. Но завести их не получилось. Поэтому я и здесь

VadimK · Март 25, 2022 11:47:14

Должно запуститься как :

 cli = CreateNotifications()
cli.run()

Celery же наверное не постгрес в качестве брокера использует. Т.е. ошибка о соединение - это куда то в другую степь. Как вариант всю работу с базой закоментировать и проверить на простых print выводах.

Python-сообщество

Уведомления

#1 Март 16, 2022 20:30:07

Как запустить celery таску в pytest

#2 Март 24, 2022 01:06:24

Как запустить celery таску в pytest

#3 Март 24, 2022 09:39:16

Как запустить celery таску в pytest

#4 Март 24, 2022 18:57:22

Как запустить celery таску в pytest

#5 Март 24, 2022 19:23:14

Как запустить celery таску в pytest

#6 Март 25, 2022 11:47:14

Как запустить celery таску в pytest

Board footer