Changeset View

Standalone View

				import os
	from os import path			from os import path

				from celery import shared_task
				from celery.contrib.testing.worker import _start_worker_thread
				from celery import current_app

				__all__ = ['start_worker_thread']

	DATA_DIR = path.join(path.dirname(__file__), 'data')			DATA_DIR = path.join(path.dirname(__file__), 'data')

				os.environ['CELERY_BROKER_URL'] = 'memory://'
				os.environ['CELERY_RESULT_BACKEND'] = 'redis://localhost'
				current_app.conf.update(
				celery_task_serializer='msgpack',
				)


				def start_worker_thread():
				return _start_worker_thread(current_app, loglevel='DEBUG')


				# Needed to pass an assertion, see
				# https://github.com/celery/celery/pull/5111
				@shared_task(name='celery.ping')
				def ping():
				return 'pong'