Added alert when job.notification_count doesn't match total notification for job

- Added log for when a job starts so that we will know when the processing of a job starts with the number of notifications
- Added dao method to get total notifications for a job id
- Added a test to check whether the number of notifications in the table matches the job notification_count

app/celery/tasks.py

@@ -29,6 +29,7 @@ from app.dao.jobs_dao import (
     dao_update_job_status)
 from app.dao.notifications_dao import (
     get_notification_by_id,
+    dao_get_total_notifications_for_job_id,
     dao_update_notifications_for_job_to_sent_to_dvla,
     dao_update_notifications_by_reference
 )
@@ -85,6 +86,8 @@ def process_job(job_id):
     TemplateClass = get_template_class(db_template.template_type)
     template = TemplateClass(db_template.__dict__)
 
+    current_app.logger.info("Starting job {} processing {} notifications".format(job_id, job.notification_count))
+
     for row_number, recipient, personalisation in RecipientCSV(
             s3.get_job_from_s3(str(service.id), str(job_id)),
             template_type=template.template_type,
@@ -92,14 +95,21 @@ def process_job(job_id):
     ).enumerated_recipients_and_personalisation:
         process_row(row_number, recipient, personalisation, template, job, service)
 
-    if template.template_type == LETTER_TYPE:
-        if service.research_mode:
-            update_job_to_sent_to_dvla.apply_async([str(job.id)], queue=QueueNames.RESEARCH_MODE)
-        else:
-            build_dvla_file.apply_async([str(job.id)], queue=QueueNames.JOBS)
-            current_app.logger.info("send job {} to build-dvla-file in the {} queue".format(job_id, QueueNames.JOBS))
+    notification_total_in_db = dao_get_total_notifications_for_job_id(job.id)
+
+    if job.notification_count != notification_total_in_db:
+        current_app.logger.error("Job {} is missing {} notifications".format(
+            job.id, notification_total_in_db - notification_total_in_db))
+        job.job_status = JOB_STATUS_ERROR
     else:
-        job.job_status = JOB_STATUS_FINISHED
+        if template.template_type == LETTER_TYPE:
+            if service.research_mode:
+                update_job_to_sent_to_dvla.apply_async([str(job.id)], queue=QueueNames.RESEARCH_MODE)
+            else:
+                build_dvla_file.apply_async([str(job.id)], queue=QueueNames.JOBS)
+                current_app.logger.info("send job {} to build-dvla-file in the {} queue".format(job_id, QueueNames.JOBS))
+        else:
+            job.job_status = JOB_STATUS_FINISHED
 
     finished = datetime.utcnow()
     job.processing_started = start

app/dao/notifications_dao.py

@@ -442,6 +442,16 @@ def get_total_sent_notifications_in_date_range(start_date, end_date, notificatio
     return result or 0
 
 
+def dao_get_total_notifications_for_job_id(job_id):
+    result = db.session.query(
+        func.count(Notification.id).label('count')
+    ).filter(
+        Notification.job_id == job_id
+    ).scalar()
+
+    return result or 0
+
+
 def is_delivery_slow_for_provider(
     sent_at,
     provider,

tests/app/celery/test_tasks.py

@@ -360,6 +360,29 @@ def test_should_process_all_sms_job(sample_job_with_placeholdered_template,
     assert job.job_status == 'finished'
 
 
+def test_should_error_log_missing_notifications(
+        sample_job_with_placeholdered_template, mocker):
+    multiple_sms = load_example_csv('multiple_sms').strip()
+    num_phone_numbers_after_header = len(multiple_sms.split('\n')[1:])
+    sample_job_with_placeholdered_template.notification_count = num_phone_numbers_after_header
+
+    mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=multiple_sms)
+    mocker.patch('app.celery.tasks.send_sms.apply_async')
+    mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
+    mocker.patch('app.celery.tasks.create_uuid', return_value="uuid")
+    # deliberately return wrong total notifications to trigger error log
+    mocker.patch(
+        'app.celery.tasks.dao_get_total_notifications_for_job_id',
+        return_value=num_phone_numbers_after_header - 1
+    )
+    error_log = mocker.patch('app.celery.tasks.current_app.logger.error')
+
+    process_job(sample_job_with_placeholdered_template.id)
+
+    job = jobs_dao.dao_get_job_by_id(sample_job_with_placeholdered_template.id)
+    assert job.job_status == 'error'
+    assert error_log.called
+
 # -------------- process_row tests -------------- #
 
 

tests/app/dao/notification_dao/test_notification_dao.py

@@ -35,6 +35,7 @@ from app.dao.notifications_dao import (
     dao_get_potential_notification_statistics_for_day,
     dao_get_scheduled_notifications,
     dao_get_template_usage,
+    dao_get_total_notifications_for_job_id,
     dao_timeout_notifications,
     dao_update_notification,
     dao_update_notifications_for_job_to_sent_to_dvla,
@@ -52,7 +53,12 @@ from app.dao.notifications_dao import (
 )
 
 from app.dao.services_dao import dao_update_service
-from tests.app.db import create_notification, create_api_key, create_reply_to_email
+from tests.app.db import (
+    create_api_key,
+    create_job,
+    create_notification,
+    create_reply_to_email
+)
 from tests.app.conftest import (
     sample_notification,
     sample_template,
@@ -1995,3 +2001,13 @@ def test_dao_get_notification_ememail_reply_toail_reply_for_notification(sample_
 
 def test_dao_get_notification_email_reply_for_notification_where_no_mapping(notify_db_session, fake_uuid):
     assert dao_get_notification_email_reply_for_notification(fake_uuid) is None
+
+
+def test_dao_get_total_notifications_for_job_id(sample_job):
+    job = create_job(sample_job.template)
+    create_notification(sample_job.template, job=sample_job)
+    create_notification(sample_job.template, job=sample_job)
+    create_notification(sample_job.template, job=sample_job)
+    create_notification(sample_job.template, job=job)
+
+    assert dao_get_total_notifications_for_job_id(sample_job.id) == 3