<s> """<STR_LIT>""" <EOL> import threading <EOL> import time <EOL> from celery . signals import worker_ready <EOL> import pytest <EOL> from tests . instances import app , celery <EOL> WORKER_READY = list ( ) <EOL> class Worker ( threading . Thread ) : <EOL> def run ( self ) : <EOL> celery_args = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:-c>' , '<STR_LIT:1>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> with app . app_context ( ) : <EOL> celery . worker_main ( celery_args ) <EOL> @ worker_ready . connect <EOL> def on_worker_ready ( ** _ ) : <EOL> """<STR_LIT>""" <EOL> WORKER_READY . append ( True ) <EOL> @ pytest . fixture ( autouse = True , scope = '<STR_LIT>' ) <EOL> def celery_worker ( ) : <EOL> """<STR_LIT>""" <EOL> thread = Worker ( ) <EOL> thread . daemon = True <EOL> thread . start ( ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> if WORKER_READY : <EOL> break <EOL> time . sleep ( <NUM_LIT:1> ) </s>
