contrib.testing.worker

Embedded workers for integration tests.

Module Contents

Classes

TestWorkController(self,*args,**kwargs) Worker that can synchronize on being fully started.

Functions

start_worker(app,concurrency=1,pool=”solo”,loglevel=WORKER_LOGLEVEL,logfile=None,perform_ping_check=True,ping_task_timeout=10.0,**kwargs) Start embedded worker.
_start_worker_thread(app,concurrency=1,pool=”solo”,loglevel=WORKER_LOGLEVEL,logfile=None,WorkController=TestWorkController,**kwargs) Start Celery worker in a thread.
_start_worker_process(app,concurrency=1,pool=”solo”,loglevel=WORKER_LOGLEVEL,logfile=None,**kwargs) Start worker in separate process.
setup_app_for_worker(app,loglevel,logfile) Setup the app to be used for starting an embedded worker.
class TestWorkController(*args, **kwargs)

Worker that can synchronize on being fully started.

__init__(*args, **kwargs)
on_consumer_ready(consumer)

Callback called when the Consumer blueprint is fully started.

ensure_started()

Wait for worker to be fully up and running.

Warning:
Worker must be started within a thread for this to work, or it will block forever.
start_worker(app, concurrency=1, pool="solo", loglevel=WORKER_LOGLEVEL, logfile=None, perform_ping_check=True, ping_task_timeout=10.0, **kwargs)

Start embedded worker.

Yields:
celery.app.worker.Worker: worker instance.
_start_worker_thread(app, concurrency=1, pool="solo", loglevel=WORKER_LOGLEVEL, logfile=None, WorkController=TestWorkController, **kwargs)

Start Celery worker in a thread.

Yields:
celery.worker.Worker: worker instance.
_start_worker_process(app, concurrency=1, pool="solo", loglevel=WORKER_LOGLEVEL, logfile=None, **kwargs)

Start worker in separate process.

Yields:
celery.app.worker.Worker: worker instance.
setup_app_for_worker(app, loglevel, logfile)

Setup the app to be used for starting an embedded worker.