celery.contrib.pytest¶

Справочник по API¶

Fixtures and testing utilities for pytest.

celery.contrib.pytest.celery_app(request, celery_config, celery_parameters, celery_enable_logging, use_celery_app_trap)[исходный код]¶

Fixture creating a Celery application instance.

celery.contrib.pytest.celery_class_tasks()[исходный код]¶

Redefine this fixture to register tasks with the test Celery app.

celery.contrib.pytest.celery_config() → Mapping[str, Any][исходный код]¶

Redefine this fixture to configure the test Celery app.

The config returned by your fixture will then be used to configure the celery_app() fixture.

celery.contrib.pytest.celery_enable_logging() → bool[исходный код]¶

You can override this fixture to enable logging.

celery.contrib.pytest.celery_includes() → Sequence[str][исходный код]¶

You can override this include modules when a worker start.

You can have this return a list of module names to import, these can be task modules, modules registering signals, and so on.

celery.contrib.pytest.celery_parameters() → Mapping[str, Any][исходный код]¶

Redefine this fixture to change the init parameters of test Celery app.

The dict returned by your fixture will then be used as parameters when instantiating Celery.

celery.contrib.pytest.celery_session_app(request: Any, celery_config: Any, celery_parameters: Any, celery_enable_logging: Any, use_celery_app_trap: Any) → object[исходный код]¶

Session Fixture: Return app for session fixtures.

celery.contrib.pytest.celery_session_worker(request: Any, celery_session_app: object, celery_includes: Sequence[str], celery_class_tasks: str, celery_worker_pool: Any, celery_worker_parameters: Mapping[str, Any]) → object[исходный код]¶

Session Fixture: Start worker that lives throughout test suite.

celery.contrib.pytest.celery_worker(request: Any, celery_app: object, celery_includes: Sequence[str], celery_worker_pool: str, celery_worker_parameters: Any) → object[исходный код]¶

Fixture: Start worker in a thread, stop it when the test returns.

celery.contrib.pytest.celery_worker_parameters() → Mapping[str, Any][исходный код]¶

Redefine this fixture to change the init parameters of Celery workers.

This can be used e. g. to define queues the worker will consume tasks from.

The dict returned by your fixture will then be used as parameters when instantiating WorkController.

celery.contrib.pytest.celery_worker_pool() → Union[str, Any][исходный код]¶

You can override this fixture to set the worker pool.

The «solo» pool is used by default, but you can set this to return e.g. «prefork».

celery.contrib.pytest.depends_on_current_app(celery_app)[исходный код]¶

Fixture that sets app as current.

celery.contrib.pytest.pytest_configure(config)[исходный код]¶

Register additional pytest configuration.

celery.contrib.pytest.use_celery_app_trap() → bool[исходный код]¶

You can override this fixture to enable the app trap.

The app trap raises an exception whenever something attempts to use the current or default apps.