Process Incomplete Jobs Updates

Comments are PR review. Updated code style in a few places to make it
more consistent with other code, added tests for letters and emails
so they are testedt, refactored some database queries to dao file

- Fixed code style
- Refactored database queries to dao code
- Added tests for emails and sms.

app/celery/scheduled_tasks.py

@@ -42,8 +42,8 @@ from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
 from app.models import (
     Job,
     LETTER_TYPE,
-    JOB_STATUS_READY_TO_SEND,
-    JOB_STATUS_IN_PROGRESS
+    JOB_STATUS_IN_PROGRESS,
+    JOB_STATUS_READY_TO_SEND
 )
 from app.notifications.process_notifications import send_notification_to_queue
 from app.statsd_decorators import statsd

app/celery/tasks.py

@@ -37,8 +37,8 @@ from app.dao.jobs_dao import (
 from app.dao.notifications_dao import (
     get_notification_by_id,
     dao_update_notifications_for_job_to_sent_to_dvla,
-    dao_update_notifications_by_reference
-)
+    dao_update_notifications_by_reference,
+    dao_get_last_notification_added_for_job_id)
 from app.dao.provider_details_dao import get_current_provider
 from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service
 from app.dao.services_dao import dao_fetch_service_by_id, fetch_todays_total_message_count
@@ -47,13 +47,13 @@ from app.models import (
     Job,
     Notification,
     EMAIL_TYPE,
-    KEY_TYPE_NORMAL,
     JOB_STATUS_CANCELLED,
-    JOB_STATUS_PENDING,
-    JOB_STATUS_IN_PROGRESS,
     JOB_STATUS_FINISHED,
+    JOB_STATUS_IN_PROGRESS,
+    JOB_STATUS_PENDING,
     JOB_STATUS_READY_TO_SEND,
     JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_ERROR,
+    KEY_TYPE_NORMAL,
     LETTER_TYPE,
     NOTIFICATION_SENDING,
     NOTIFICATION_TECHNICAL_FAILURE,
@@ -109,11 +109,11 @@ def process_job(job_id):
     ).enumerated_recipients_and_personalisation:
         process_row(row_number, recipient, personalisation, template, job, service)
 
-    job_complete(job, service, template, False, start)
+    job_complete(job, service, template.template_type, start=start)
 
 
-def job_complete(job, service, template, resumed, start=None):
-    if template.template_type == LETTER_TYPE:
+def job_complete(job, service, template_type, resumed=False, start=None):
+    if template_type == LETTER_TYPE:
         if service.research_mode:
             update_job_to_sent_to_dvla.apply_async([str(job.id)], queue=QueueNames.RESEARCH_MODE)
         else:
@@ -506,13 +506,9 @@ def process_incomplete_jobs(job_ids):
 
 def process_incomplete_job(job_id):
 
-    job = Job.query.filter(Job.id == job_id).one()
+    job = dao_get_job_by_id(job_id)
 
-    last_notification_added = Notification.query.filter(
-        Notification.job_id == job_id
-    ).order_by(
-        Notification.job_row_number.desc()
-    ).first()
+    last_notification_added = dao_get_last_notification_added_for_job_id(job_id)
 
     if last_notification_added:
         resume_from_row = last_notification_added.job_row_number
@@ -534,4 +530,4 @@ def process_incomplete_job(job_id):
         if row_number > resume_from_row:
             process_row(row_number, recipient, personalisation, template, job, job.service)
 
-    job_complete(job, job.service, template, True)
+    job_complete(job, job.service, template, resumed=True)

app/dao/notifications_dao.py

@@ -640,3 +640,14 @@ def dao_get_notification_email_reply_for_notification(notification_id):
 
     if email_reply_to:
         return email_reply_to.email_address
+
+
+@statsd(namespace="dao")
+def dao_get_last_notification_added_for_job_id(job_id):
+    last_notification_added = Notification.query.filter(
+        Notification.job_id == job_id
+    ).order_by(
+        Notification.job_row_number.desc()
+    ).first()
+
+    return last_notification_added

test_csv_files/multiple_letter.csv

@@ -0,0 +1,11 @@
+address_line_1, address_line_2, address_line_3
+name1, street1, town1, postcode1
+name2, street2, town2, postcode2
+name3, street3, town3, postcode3
+name4, street4, town4, postcode4
+name5, street5, town5, postcode5
+name6, street6, town6, postcode6
+name7, street7, town7, postcode7
+name8, street8, town8, postcode8
+name9, street9, town9, postcode9
+name0, street0, town0, postcode0

tests/app/celery/test_tasks.py

@@ -34,11 +34,11 @@ from app.models import (
     Job,
     Notification,
     EMAIL_TYPE,
-    JOB_STATUS_FINISHED,
-    JOB_STATUS_IN_PROGRESS,
     KEY_TYPE_NORMAL,
     KEY_TYPE_TEAM,
     KEY_TYPE_TEST,
+    JOB_STATUS_FINISHED,
+    JOB_STATUS_IN_PROGRESS,
     LETTER_TYPE,
     SERVICE_PERMISSION_TYPES,
     SMS_TYPE
@@ -1226,12 +1226,12 @@ def test_check_job_status_task_does_not_raise_error(sample_template):
     check_job_status()
 
 
-def test_process_incomplete_job(mocker, sample_template):
+def test_process_incomplete_job_sms(mocker, sample_template):
 
     mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
     send_sms = mocker.patch('app.celery.tasks.send_sms.apply_async')
 
-    job = create_job(template=sample_template, notification_count=3,
+    job = create_job(template=sample_template, notification_count=10,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
@@ -1256,7 +1256,7 @@ def test_process_incomplete_job_with_notifications_all_sent(mocker, sample_templ
     mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
     mock_send_sms = mocker.patch('app.celery.tasks.send_sms.apply_async')
 
-    job = create_job(template=sample_template, notification_count=3,
+    job = create_job(template=sample_template, notification_count=10,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
@@ -1281,15 +1281,15 @@ def test_process_incomplete_job_with_notifications_all_sent(mocker, sample_templ
 
     assert completed_job.job_status == JOB_STATUS_FINISHED
 
-    assert mock_send_sms.call_count == 0  # There are 10 in the file and we've added two already
+    assert mock_send_sms.call_count == 0  # There are 10 in the file and we've added 10 it should not have been called
 
 
-def test_process_incomplete_jobs(mocker, sample_template):
+def test_process_incomplete_jobs_sms(mocker, sample_template):
 
     mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
     mock_send_sms = mocker.patch('app.celery.tasks.send_sms.apply_async')
 
-    job = create_job(template=sample_template, notification_count=3,
+    job = create_job(template=sample_template, notification_count=10,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
@@ -1300,7 +1300,7 @@ def test_process_incomplete_jobs(mocker, sample_template):
 
     assert Notification.query.filter(Notification.job_id == job.id).count() == 3
 
-    job2 = create_job(template=sample_template, notification_count=3,
+    job2 = create_job(template=sample_template, notification_count=10,
                       created_at=datetime.utcnow() - timedelta(hours=2),
                       scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                       processing_started=datetime.utcnow() - timedelta(minutes=31),
@@ -1331,7 +1331,7 @@ def test_process_incomplete_jobs_no_notifications_added(mocker, sample_template)
     mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
     mock_send_sms = mocker.patch('app.celery.tasks.send_sms.apply_async')
 
-    job = create_job(template=sample_template, notification_count=3,
+    job = create_job(template=sample_template, notification_count=10,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
@@ -1356,7 +1356,7 @@ def test_process_incomplete_jobs(mocker):
     jobs = []
     process_incomplete_jobs(jobs)
 
-    assert mock_send_sms.call_count == 0  # There are 20 in total over 2 jobs we've added 8 already
+    assert mock_send_sms.call_count == 0  # There are no jobs to process so it will not have been called
 
 
 def test_process_incomplete_job_no_job_in_database(mocker, fake_uuid):
@@ -1367,4 +1367,54 @@ def test_process_incomplete_job_no_job_in_database(mocker, fake_uuid):
     with pytest.raises(expected_exception=Exception) as e:
         process_incomplete_job(fake_uuid)
 
-    assert mock_send_sms.call_count == 0  # There are 20 in total over 2 jobs we've added 8 already
+    assert mock_send_sms.call_count == 0  # There is no job in the db it will not have been called
+
+
+def test_process_incomplete_job_email(mocker, sample_email_template):
+
+    mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_email'))
+    mock_email_sender = mocker.patch('app.celery.tasks.send_email.apply_async')
+
+    job = create_job(template=sample_email_template, notification_count=10,
+                     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)
+
+    create_notification(sample_email_template, job, 0)
+    create_notification(sample_email_template, job, 1)
+
+    assert Notification.query.filter(Notification.job_id == job.id).count() == 2
+
+    process_incomplete_job(str(job.id))
+
+    completed_job = Job.query.filter(Job.id == job.id).one()
+
+    assert completed_job.job_status == JOB_STATUS_FINISHED
+
+    assert mock_email_sender.call_count == 8  # There are 10 in the file and we've added two already
+
+
+def test_process_incomplete_job_letter(mocker, sample_letter_template):
+
+    mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_letter'))
+    mock_letter_sender = mocker.patch('app.celery.tasks.persist_letter.apply_async')
+
+    job = create_job(template=sample_letter_template, notification_count=10,
+                     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)
+
+    create_notification(sample_letter_template, job, 0)
+    create_notification(sample_letter_template, job, 1)
+
+    assert Notification.query.filter(Notification.job_id == job.id).count() == 2
+
+    process_incomplete_job(str(job.id))
+
+    completed_job = Job.query.filter(Job.id == job.id).one()
+
+    assert completed_job.job_status == JOB_STATUS_FINISHED
+
+    assert mock_letter_sender.call_count == 8

tests/app/dao/notification_dao/test_notification_dao.py

@@ -22,8 +22,8 @@ from app.models import (
     NOTIFICATION_DELIVERED,
     KEY_TYPE_NORMAL,
     KEY_TYPE_TEAM,
-    KEY_TYPE_TEST
-)
+    KEY_TYPE_TEST,
+    JOB_STATUS_IN_PROGRESS)
 
 from app.dao.notifications_dao import (
     dao_create_notification,
@@ -50,8 +50,8 @@ from app.dao.notifications_dao import (
     is_delivery_slow_for_provider,
     set_scheduled_notification_to_processed,
     update_notification_status_by_id,
-    update_notification_status_by_reference
-)
+    update_notification_status_by_reference,
+    dao_get_last_notification_added_for_job_id)
 
 from app.dao.services_dao import dao_update_service
 from tests.app.db import (
@@ -2037,3 +2037,31 @@ def test_dao_get_notification_email_reply_for_notification(sample_service, sampl
 
 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_last_notification_added_for_job_id_valid_job_id(sample_template):
+    job = create_job(template=sample_template, notification_count=10,
+                     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)
+    create_notification(sample_template, job, 0)
+    create_notification(sample_template, job, 1)
+    last = create_notification(sample_template, job, 2)
+
+    assert dao_get_last_notification_added_for_job_id(job.id) == last
+
+
+def test_dao_get_last_notification_added_for_job_id_no_notifications(sample_template):
+    job = create_job(template=sample_template, notification_count=10,
+                     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)
+
+    assert dao_get_last_notification_added_for_job_id(job.id) is None
+
+
+def test_dao_get_last_notification_added_for_job_id_no_notifications(sample_template, fake_uuid):
+
+    assert dao_get_last_notification_added_for_job_id(fake_uuid) is None