celery.contrib.testing.mocks

Полезные моки для модульного тестирования.

celery.contrib.testing.mocks.ContextMock(*args, **kwargs)

Мок для контекстов операторов with.

celery.contrib.testing.mocks.TaskMessage(name:str, id:str|None=None, args:Sequence=(), kwargs:Mapping|None=None, callbacks:Sequence[Signature]|None=None, errbacks:Sequence[Signature]|None=None, chain:Sequence[Signature]|None=None, shadow:str|None=None, utc:bool|None=None, **options:Any) → Any

Создать сообщение задачи в формате протокола 2.

celery.contrib.testing.mocks.TaskMessage1(name:str, id:str|None=None, args:Sequence=(), kwargs:Mapping|None=None, callbacks:Sequence[Signature]|None=None, errbacks:Sequence[Signature]|None=None, chain:Sequence[Signature]|None=None, **options:Any) → Any

Создать сообщение задачи в формате протокола 1.

celery.contrib.testing.mocks.task_message_from_sig(app: ~celery.app.base.Celery, sig: ~celery.canvas.Signature, utc: bool = True, TaskMessage: ~typing.Any = <function TaskMessage>) → Any

Создать сообщение задачи из celery.Signature.

Пример

>>> m = task_message_from_sig(app, add.s(2, 2))
>>> amqp_client.basic_publish(m, exchange='ex', routing_key='rkey')