cic-internal-integration/apps/cic-ussd/tests/fixtures/tasker.py

58 lines
1.3 KiB
Python
Raw Permalink Normal View History

2021-02-06 16:13:47 +01:00
# standard imports
import logging
2021-08-06 18:29:01 +02:00
import os
2021-02-06 16:13:47 +01:00
import pytest
import shutil
import tempfile
2021-08-06 18:29:01 +02:00
# external imports
from celery import uuid
2021-02-06 16:13:47 +01:00
logg = logging.getLogger()
@pytest.fixture(scope='session')
def celery_includes():
return [
'cic_ussd.tasks.callback_handler',
2021-08-06 18:29:01 +02:00
'cic_ussd.tasks.metadata',
'cic_ussd.tasks.notifications',
'cic_ussd.tasks.processor',
'cic_ussd.tasks.ussd_session',
'cic_eth.queue.balance',
'cic_notify.tasks.sms',
2021-02-06 16:13:47 +01:00
]
@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_enable_logging():
return True
2021-08-06 18:29:01 +02:00
@pytest.fixture(scope='function')
def task_uuid():
return uuid()