tests/conftest.py
def director_worker(request):
    """Fixture starting a celery worker in background"""

    from multiprocessing import Process
    from harvester.director.tasks import worker
    p = Process(target=lambda: worker.worker_main(['test-celery-worker']))

    def cleanup():
        p.terminate()

    request.addfinalizer(cleanup)
    p.start()
    return p
Fixture starting a celery worker in background
Fixture starting a celery worker in background
Fixture starting a celery worker in background
def director_worker(request):
    from multiprocessing import Process
    from harvester.director.tasks import worker
    p = Process(target=lambda: worker.worker_main(['test-celery-worker']))
    def cleanup():
        p.terminate()
    request.addfinalizer(cleanup)
    p.start()
    return p
def director_worker ( request ) : from multiprocessing import Process from harvester . director . tasks import worker p = Process ( target = lambda : worker . worker_main ( [ 'test-celery-worker' ] ) ) def cleanup ( ) : p . terminate ( ) request . addfinalizer ( cleanup ) p . start ( ) return p
Fixture starting a celery worker in background
Fixture starting a celery worker in background
"""Fixture starting a celery worker in background"""
request