49 lines
1.1 KiB
Python
49 lines
1.1 KiB
Python
|
# third-party imports
|
|||
|
|
import pytest
|
|||
|
|
import tempfile
|
|||
|
|
import logging
|
|||
|
|
import shutil
|
|||
|
|
|
|||
|
|
logg = logging.getLogger(__file__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# celery fixtures
|
|||
|
|
@pytest.fixture(scope='session')
|
|||
|
|
def celery_includes():
|
|||
|
|
return [
|
|||
|
|
'cic_notify.tasks.sms',
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope='session')
|
|||
|
|
def celery_config():
|
|||
|
|
bq = tempfile.mkdtemp()
|
|||
|
|
bp = tempfile.mkdtemp()
|
|||
|
|
rq = tempfile.mkdtemp()
|
|||
|
|
logg.debug('celery broker queue {} processed {}'.format(bq, bp))
|
|||
|
|
logg.debug('celery backend store {}'.format(rq))
|
|||
|
|
yield {
|
|||
|
|
'broker_url': 'filesystem://',
|
|||
|
|
'broker_transport_options': {
|
|||
|
|
'data_folder_in': bq,
|
|||
|
|
'data_folder_out': bq,
|
|||
|
|
'data_folder_processed': bp,
|
|||
|
|
},
|
|||
|
|
'result_backend': 'file://{}'.format(rq),
|
|||
|
|
}
|
|||
|
|
logg.debug('cleaning up celery filesystem backend files {} {} {}'.format(bq, bp, rq))
|
|||
|
|
shutil.rmtree(bq)
|
|||
|
|
shutil.rmtree(bp)
|
|||
|
|
shutil.rmtree(rq)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope='session')
|
|||
|
|
def celery_worker_parameters():
|
|||
|
|
return {
|
|||
|
|
# 'queues': ('cic-notify'),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope='session')
|
|||
|
|
def celery_enable_logging():
|
|||
|
|
return True
|