Finished celery refactor - set up config for queue prefix

LEO notes: Also made sure the Test BROKER_URL is preserved so that
tests warn you when celery isn't mocked out
This commit is contained in:
Martyn Inglis
2017-06-09 16:20:02 +01:00
committed by Leo Hemsted
parent 786adb5d71
commit 0e9e8955f7
4 changed files with 33 additions and 20 deletions

View File

@@ -49,7 +49,6 @@ def create_app(app_name=None):
from app.config import configs
notify_environment = os.environ['NOTIFY_ENVIRONMENT']
application.config.from_object(configs[notify_environment])
if app_name:

View File

@@ -7,14 +7,16 @@ from kombu import Queue, Exchange
from app.config import QueueNames
# BROKER_URL = 'you-forgot-to-mock-celery-in-your-tests://'
class CeleryConfig(object):
broker_url = 'sqs://'
class CeleryConfig:
def __init__(self, config):
self.broker_transport_options['queue_name_prefix'] = config['NOTIFICATION_QUEUE_PREFIX']
self.broker_url = config.get('BROKER_URL', 'sqs://')
broker_transport_options = {
'region': 'sqs.eu-west-1',
'polling_interval': 1, # 1 second
'visibility_timeout': 310,
'queue_name_prefix': 'martyn-'
'queue_name_prefix': None
}
enable_utc = True,
timezone = 'Europe/London'
@@ -110,9 +112,17 @@ class CeleryConfig(object):
class NotifyCelery(Celery):
def init_app(self, app):
super().__init__(app.import_name, broker=CeleryConfig.broker_url)
self.init_queues_if_needed(app.config['NOTIFY_ENVIRONMENT'])
self.config_from_object(CeleryConfig())
celery_config = CeleryConfig(app.config)
super().__init__(app.import_name, broker=celery_config.broker_url)
if app.config['INITIALISE_QUEUES']:
for queue in QueueNames.all_queues():
CeleryConfig.task_queues.append(
Queue(queue, Exchange('default'), routing_key=queue)
)
self.config_from_object()
TaskBase = self.Task
class ContextTask(TaskBase):
@@ -123,10 +133,3 @@ class NotifyCelery(Celery):
return TaskBase.__call__(self, *args, **kwargs)
self.Task = ContextTask
def init_queues_if_needed(self, environment):
if environment in ['development', 'test']:
for queue in QueueNames.all_queues():
CeleryConfig.task_queues.append(
Queue(queue, Exchange('default'), routing_key=queue)
)

View File

@@ -132,6 +132,7 @@ class Config(object):
}
FREE_SMS_TIER_FRAGMENT_COUNT = 250000
INITIALISE_QUEUES = False
SMS_INBOUND_WHITELIST = json.loads(os.environ.get('SMS_INBOUND_WHITELIST', '[]'))
@@ -141,12 +142,12 @@ class Config(object):
######################
class Development(Config):
INITIALISE_QUEUES = True
SQLALCHEMY_ECHO = False
NOTIFY_EMAIL_DOMAIN = 'notify.tools'
CSV_UPLOAD_BUCKET_NAME = 'development-notifications-csv-upload'
DVLA_RESPONSE_BUCKET_NAME = 'notify.tools-ftp'
NOTIFY_ENVIRONMENT = 'development'
NOTIFICATION_QUEUE_PREFIX = 'development'
DEBUG = True
API_HOST_NAME = "http://localhost:6011"
@@ -154,6 +155,7 @@ class Development(Config):
class Test(Config):
INITIALISE_QUEUES = True
NOTIFY_EMAIL_DOMAIN = 'test.notify.com'
FROM_NUMBER = 'testing'
NOTIFY_ENVIRONMENT = 'test'
@@ -165,6 +167,8 @@ class Test(Config):
STATSD_HOST = "localhost"
STATSD_PORT = 1000
BROKER_URL = 'you-forgot-to-mock-celery-in-your-tests://'
API_RATE_LIMIT_ENABLED = True
API_HOST_NAME = "http://localhost:6011"

View File

@@ -1,14 +1,21 @@
import pytest
from sqlalchemy.exc import SQLAlchemyError
from app import create_uuid
from app.celery.statistics_tasks import (
record_initial_job_statistics,
record_outcome_job_statistics,
create_initial_notification_statistic_tasks,
create_outcome_notification_statistic_tasks)
from sqlalchemy.exc import SQLAlchemyError
from app import create_uuid
from app.models import (
NOTIFICATION_STATUS_TYPES_COMPLETED,
NOTIFICATION_SENDING,
NOTIFICATION_PENDING,
NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED
)
from tests.app.conftest import sample_notification
from app.models import NOTIFICATION_STATUS_TYPES_COMPLETED, NOTIFICATION_SENT, NOTIFICATION_SENDING, \
NOTIFICATION_PENDING, NOTIFICATION_CREATED, NOTIFICATION_DELIVERED
def test_should_create_initial_job_task_if_notification_is_related_to_a_job(