Function start_worker

celery/contrib/testing/worker.py:100–136

Start embedded worker. Yields: celery.app.worker.Worker: worker instance.

(
    app,  # type: Celery
    concurrency=1,  # type: int
    pool='solo',  # type: str
    loglevel=WORKER_LOGLEVEL,  # type: Union[str, int]
    logfile=None,  # type: str
    perform_ping_check=True,  # type: bool
    ping_task_timeout=10.0,  # type: float
    shutdown_timeout=10.0,  # type: float
    **kwargs  # type: Any
)

@contextmanager
def start_worker(
    app,  # type: Celery
    concurrency=1,  # type: int
    pool='solo',  # type: str
    loglevel=WORKER_LOGLEVEL,  # type: Union[str, int]
    logfile=None,  # type: str
    perform_ping_check=True,  # type: bool
    ping_task_timeout=10.0,  # type: float
    shutdown_timeout=10.0,  # type: float
    **kwargs  # type: Any
):
    # type: (...) -> Iterable
    """Start embedded worker.

    Yields:
        celery.app.worker.Worker: worker instance.
    """
    test_worker_starting.send(sender=app)

    worker = None
    try:
        with _start_worker_thread(app,
                                  concurrency=concurrency,
                                  pool=pool,
                                  loglevel=loglevel,
                                  logfile=logfile,
                                  perform_ping_check=perform_ping_check,
                                  shutdown_timeout=shutdown_timeout,
                                  **kwargs) as worker:
            if perform_ping_check:
                from .tasks import ping
                with allow_join_result():
                    assert ping.delay().get(timeout=ping_task_timeout) == 'pong'

            yield worker
    finally:
        test_worker_stopped.send(sender=app, worker=worker)


@contextmanager