darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-11 09:27:56 -04:00
Code Issues Packages Projects Releases Wiki Activity

Added a new scheduled task that runs every minute, may want to run it every 3 minutes.

Browse Source 
The tasks checks that a job is not still in progress 30 minutes after the job started processing.
This commit is contained in:
Rebecca Law
2017-10-12 16:21:08 +01:00
parent c068c922b4
commit e08690cad2
5 changed files with 104 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
app/celery/scheduled_tasks.py
View File

@@ -4,6 +4,7 @@ from datetime import (
)
from flask import current_app
from sqlalchemy import or_, and_
from sqlalchemy.exc import SQLAlchemyError
from notifications_utils.s3 import s3upload
@@ -17,8 +18,8 @@ from app.dao.invited_user_dao import delete_invitations_created_more_than_two_da
from app.dao.jobs_dao import (
dao_get_letter_job_ids_by_status,
dao_set_scheduled_jobs_to_pending,
dao_get_jobs_older_than_limited_by
)
dao_get_jobs_older_than_limited_by,
dao_get_job_by_id)
from app.dao.monthly_billing_dao import (
get_service_ids_that_need_billing_populated,
create_or_update_monthly_billing
@@ -37,12 +38,14 @@ from app.dao.provider_details_dao import (
dao_toggle_sms_provider
)
from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
from app.models import LETTER_TYPE, JOB_STATUS_READY_TO_SEND
from app.models import LETTER_TYPE, JOB_STATUS_READY_TO_SEND, JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_FINISHED, Job, \
EMAIL_TYPE, SMS_TYPE, JOB_STATUS_IN_PROGRESS
from app.notifications.process_notifications import send_notification_to_queue
from app.statsd_decorators import statsd
from app.celery.tasks import process_job, create_dvla_file_contents_for_notifications
from app.config import QueueNames, TaskNames
from app.utils import convert_utc_to_bst
from app.v2.errors import JobIncompleteError
@notify_celery.task(name="remove_csv_files")
@@ -354,3 +357,30 @@ def run_letter_api_notifications():
QueueNames.PROCESS_FTP
)
)
@notify_celery.task(name='check-job-status')
@statsd(namespace="tasks")
def check_job_status():
"""
every x minutes do this check
select
from jobs
where job_status == 'in progress'
and template_type in ('sms', 'email')
and scheduled_at or created_at is older that 30 minutes.
if any results then
raise error
process the rows in the csv thatgit c are missing (in another task) just do the check here.
"""
thirty_minutes_ago = datetime.utcnow() - timedelta(minutes=30)
thirty_five_minutes_ago = datetime.utcnow() - timedelta(minutes=35)
jobs_not_complete_after_30_minutes = Job.query.filter(
Job.job_status == JOB_STATUS_IN_PROGRESS,
and_(thirty_five_minutes_ago < Job.processing_started, Job.processing_started < thirty_minutes_ago)
).all()
job_ids = [str(x.id) for x in jobs_not_complete_after_30_minutes]
if job_ids:
raise JobIncompleteError("Job(s) {} have not completed.".format(job_ids))

27
app/celery/tasks.py
View File

@@ -6,9 +6,15 @@ from flask import current_app
from notifications_utils.recipients import (
RecipientCSV
)
from notifications_utils.template import SMSMessageTemplate, WithSubjectTemplate, LetterDVLATemplate
from requests import HTTPError
from requests import request
from notifications_utils.template import (
SMSMessageTemplate,
WithSubjectTemplate,
LetterDVLATemplate
)
from requests import (
HTTPError,
request
)
from sqlalchemy.exc import SQLAlchemyError
from app import (
create_uuid,
@@ -53,7 +59,6 @@ from app.models import (
from app.notifications.process_notifications import persist_notification
from app.service.utils import service_allowed_to_send_to
from app.statsd_decorators import statsd
from app.v2.errors import JobIncompleteError
from notifications_utils.s3 import s3upload
@@ -79,6 +84,7 @@ def process_job(job_id):
return
job.job_status = JOB_STATUS_IN_PROGRESS
job.processing_started = start
dao_update_job(job)
db_template = dao_get_template_by_id(job.template_id, job.template_version)
@@ -88,8 +94,6 @@ def process_job(job_id):
current_app.logger.info("Starting job {} processing {} notifications".format(job_id, job.notification_count))
check_job_status.apply_async([str(job.id)], queue=QueueNames.JOBS)
for row_number, recipient, personalisation in RecipientCSV(
s3.get_job_from_s3(str(service.id), str(job_id)),
template_type=template.template_type,
@@ -107,7 +111,6 @@ def process_job(job_id):
job.job_status = JOB_STATUS_FINISHED
finished = datetime.utcnow()
job.processing_started = start
job.processing_finished = finished
dao_update_job(job)
current_app.logger.info(
@@ -328,16 +331,6 @@ def update_dvla_job_to_error(self, job_id):
current_app.logger.info("Updated {} job to {}".format(job_id, JOB_STATUS_ERROR))
@notify_celery.task(bind=True, name='check-job-status', countdown=3600)
@statsd(namespace="tasks")
def check_job_status(self, job_id):
job = dao_get_job_by_id(job_id)
if (job.template.template_type == LETTER_TYPE and job.job_status != JOB_STATUS_SENT_TO_DVLA) or\
(job.template.template_type != LETTER_TYPE and job.job_status != JOB_STATUS_FINISHED):
raise JobIncompleteError("Job {} did not complete".format(job_id))
@notify_celery.task(bind=True, name='update-letter-notifications-to-sent')
@statsd(namespace="tasks")
def update_letter_notifications_to_sent_to_dvla(self, notification_references):

5
app/config.py
View File

@@ -236,6 +236,11 @@ class Config(object):
'task': 'run-letter-api-notifications',
'schedule': crontab(hour=17, minute=40),
'options': {'queue': QueueNames.PERIODIC}
},
'check-job-status': {
'task': 'check-job-status',
'schedule': crontab(),
'options': {'queue': QueueNames.PERIODIC}
}
}
CELERY_QUEUES = []

60
tests/app/celery/test_scheduled_tasks.py
View File

@@ -29,7 +29,7 @@ from app.celery.scheduled_tasks import (
timeout_job_statistics,
timeout_notifications,
populate_monthly_billing,
send_total_sent_notifications_to_performance_platform)
send_total_sent_notifications_to_performance_platform, check_job_status)
from app.clients.performance_platform.performance_platform_client import PerformancePlatformClient
from app.config import QueueNames, TaskNames
from app.dao.jobs_dao import dao_get_job_by_id
@@ -47,8 +47,9 @@ from app.models import (
NOTIFICATION_PENDING,
NOTIFICATION_CREATED,
KEY_TYPE_TEST,
MonthlyBilling)
MonthlyBilling, JOB_STATUS_FINISHED)
from app.utils import get_london_midnight_in_utc
from app.v2.errors import JobIncompleteError
from tests.app.db import create_notification, create_service, create_template, create_job, create_rate
from tests.app.conftest import (
sample_job as create_sample_job,
@@ -743,7 +744,6 @@ def test_run_letter_api_notifications_triggers_ftp_task(client, mocker, sample_l
def test_run_letter_api_notifications_does_nothing_if_no_created_notifications(
client,
mocker,
sample_letter_template,
sample_letter_job,
@@ -753,7 +753,7 @@ def test_run_letter_api_notifications_does_nothing_if_no_created_notifications(
sample_letter_template,
job=sample_letter_job
)
pending_letter_notification = create_notification(
create_notification(
sample_letter_template,
status=NOTIFICATION_PENDING,
api_key=sample_api_key
@@ -770,3 +770,55 @@ def test_run_letter_api_notifications_does_nothing_if_no_created_notifications(
assert not mock_celery.called
assert letter_job_notification.status == NOTIFICATION_CREATED
assert test_api_key_notification.status == NOTIFICATION_CREATED
def test_check_job_status_task_raises_job_incomplete_error(sample_template):
job = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS)
create_notification(template=sample_template, job=job)
with pytest.raises(expected_exception=JobIncompleteError) as e:
check_job_status()
assert e.value.message == "Job(s) ['{}'] have not completed.".format(str(job.id))
def test_check_job_status_task_raises_job_incomplete_error_when_scheduled_job_is_not_complete(sample_template):
job = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS)
with pytest.raises(expected_exception=JobIncompleteError) as e:
check_job_status()
assert e.value.message == "Job(s) ['{}'] have not completed.".format(str(job.id))
def test_check_job_status_task_raises_job_incomplete_error_for_multiple_jobs(sample_template):
job = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS)
job_2 = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS)
with pytest.raises(expected_exception=JobIncompleteError) as e:
check_job_status()
assert str(job.id) in e.value.message
assert str(job_2.id) in e.value.message
def test_check_job_status_task_does_not_raise_error(sample_template):
job = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_FINISHED)
job_2 = create_job(template=sample_template, notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_FINISHED)
check_job_status()

42
tests/app/celery/test_tasks.py
View File

@@ -16,9 +16,7 @@ from app.celery import tasks
from app.celery.tasks import (
s3,
build_dvla_file,
check_job_status,
create_dvla_file_contents_for_job,
update_dvla_job_to_error,
process_job,
process_row,
send_sms,
@@ -30,9 +28,6 @@ from app.config import QueueNames
from app.dao import jobs_dao, services_dao
from app.models import (
EMAIL_TYPE,
JOB_STATUS_ERROR,
JOB_STATUS_FINISHED,
JOB_STATUS_SENT_TO_DVLA,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
@@ -42,7 +37,6 @@ from app.models import (
Job,
Notification
)
from app.v2.errors import JobIncompleteError
from tests.app import load_example_csv
from tests.app.conftest import (
@@ -100,7 +94,6 @@ def email_job_with_placeholders(notify_db, notify_db_session, sample_email_templ
@freeze_time("2016-01-01 11:09:00.061258")
def test_should_process_sms_job(sample_job, mocker):
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('sms'))
mocker.patch('app.celery.tasks.send_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
@@ -137,7 +130,6 @@ def test_should_not_process_sms_job_if_would_exceed_send_limits(notify_db,
service = create_sample_service(notify_db, notify_db_session, limit=9)
job = create_sample_job(notify_db, notify_db_session, service=service, notification_count=10)
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.build_dvla_file')
@@ -159,7 +151,6 @@ def test_should_not_process_sms_job_if_would_exceed_send_limits_inc_today(notify
create_sample_notification(notify_db, notify_db_session, service=service, job=job)
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('sms'))
mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.build_dvla_file')
@@ -180,7 +171,6 @@ def test_should_not_process_email_job_if_would_exceed_send_limits_inc_today(noti
create_sample_notification(notify_db, notify_db_session, service=service, job=job)
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3')
mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.build_dvla_file')
@@ -200,7 +190,6 @@ def test_should_not_process_email_job_if_would_exceed_send_limits(notify_db, not
template = create_sample_email_template(notify_db, notify_db_session, service=service)
job = create_sample_job(notify_db, notify_db_session, service=service, template=template)
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3')
mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.build_dvla_file')
@@ -217,7 +206,6 @@ def test_should_not_process_email_job_if_would_exceed_send_limits(notify_db, not
def test_should_not_process_job_if_already_pending(notify_db, notify_db_session, mocker):
job = create_sample_job(notify_db, notify_db_session, job_status='scheduled')
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3')
mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.build_dvla_file')
@@ -237,7 +225,6 @@ def test_should_process_email_job_if_exactly_on_send_limits(notify_db,
template = create_sample_email_template(notify_db, notify_db_session, service=service)
job = create_sample_job(notify_db, notify_db_session, service=service, template=template, notification_count=10)
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_email'))
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
@@ -265,7 +252,6 @@ def test_should_process_email_job_if_exactly_on_send_limits(notify_db,
def test_should_not_create_send_task_for_empty_file(sample_job, mocker):
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('empty'))
mocker.patch('app.celery.tasks.send_sms.apply_async')
mocker.patch('app.celery.tasks.check_job_status.apply_async')
process_job(sample_job.id)
@@ -283,7 +269,6 @@ def test_should_process_email_job(email_job_with_placeholders, mocker):
email_csv = """email_address,name
test@test.com,foo
"""
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=email_csv)
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
@@ -322,7 +307,6 @@ def test_should_process_letter_job(sample_letter_job, mocker):
process_row_mock = mocker.patch('app.celery.tasks.process_row')
mocker.patch('app.celery.tasks.create_uuid', return_value="uuid")
mocker.patch('app.celery.tasks.build_dvla_file')
mocker.patch('app.celery.tasks.check_job_status.apply_async')
process_job(sample_letter_job.id)
@@ -353,7 +337,6 @@ def test_should_process_letter_job(sample_letter_job, mocker):
def test_should_process_all_sms_job(sample_job_with_placeholdered_template,
mocker):
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
mocker.patch('app.celery.tasks.send_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
@@ -458,7 +441,6 @@ def test_should_put_send_sms_task_in_research_mode_queue_if_research_mode_servic
notification = _notification_json(template, to="+447234123123")
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocked_deliver_sms = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notification_id = uuid.uuid4()
@@ -641,7 +623,6 @@ def test_should_not_build_dvla_file_in_research_mode_for_letter_job(
csv = """address_line_1,address_line_2,address_line_3,address_line_4,postcode,name
A1,A2,A3,A4,A_POST,Alice
"""
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=csv)
mocker.patch('app.celery.tasks.update_job_to_sent_to_dvla.apply_async')
mocker.patch('app.celery.tasks.persist_letter.apply_async')
@@ -664,7 +645,6 @@ def test_should_update_job_to_sent_to_dvla_in_research_mode_for_letter_job(
csv = """address_line_1,address_line_2,address_line_3,address_line_4,postcode,name
A1,A2,A3,A4,A_POST,Alice
"""
mocker.patch('app.celery.tasks.check_job_status.apply_async')
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=csv)
mock_update_job_task = mocker.patch('app.celery.tasks.update_job_to_sent_to_dvla.apply_async')
mocker.patch('app.celery.tasks.persist_letter.apply_async')
@@ -1224,25 +1204,3 @@ def test_send_inbound_sms_to_service_does_not_retries_if_request_returns_404(not
send_inbound_sms_to_service(inbound_sms.id, inbound_sms.service_id)
mocked.call_count == 0
def test_job_incomplete_raises_job_incomplete_error(sample_job):
assert sample_job.job_status != JOB_STATUS_FINISHED
with pytest.raises(JobIncompleteError) as e:
check_job_status(str(sample_job.id))
assert e.value.status_code == 500
assert e.value.message == 'Job {} did not complete'.format(sample_job.id)
@pytest.mark.parametrize('job_status,notification_type',
[
(JOB_STATUS_FINISHED, SMS_TYPE),
(JOB_STATUS_SENT_TO_DVLA, LETTER_TYPE)
]
)
def test_job_finished_does_not_raises_job_incomplete_error(
sample_job, job_status, notification_type):
sample_job.job_status = job_status
sample_job.template.template_type = notification_type
check_job_status(str(sample_job.id))