Merge pull request #2639 from alphagov/remove-loadtesting-db-migration

remove loadtesting from the database

app/celery/scheduled_tasks.py

@@ -13,7 +13,7 @@ from app.celery.tasks import process_job
 from app.config import QueueNames, TaskNames
 from app.dao.invited_org_user_dao import delete_org_invitations_created_more_than_two_days_ago
 from app.dao.invited_user_dao import delete_invitations_created_more_than_two_days_ago
-from app.dao.jobs_dao import dao_set_scheduled_jobs_to_pending
+from app.dao.jobs_dao import dao_set_scheduled_jobs_to_pending, find_jobs_with_missing_rows, find_missing_row_for_job
 from app.dao.jobs_dao import dao_update_job
 from app.dao.notifications_dao import (
     is_delivery_slow_for_provider,
@@ -222,3 +222,27 @@ def check_templated_letter_state():
                 message=msg,
                 ticket_type=zendesk_client.TYPE_INCIDENT
             )
+
+@notify_celery.task(name='find-missing-rows-from-completed-jobs')
+def find_missing_rows_from_completed_jobs():
+    jobs_and_job_size = find_jobs_with_missing_rows()
+    for x in jobs_and_job_size:
+        missing_rows = find_missing_row_for_job(jobs_and_job_size.job_id, jobs_and_job_size.notification_count)
+
+    # job = dao_get_job_by_id(job_id)
+    # db_template = dao_get_template_by_id(job.template_id, job.template_version)
+    #
+    # TemplateClass = get_template_class(db_template.template_type)
+    # template = TemplateClass(db_template.__dict__)
+    #
+    # for row in RecipientCSV(
+    #         s3.get_job_from_s3(str(job.service_id), str(job.id)),
+    #         template_type=template.template_type,
+    #         placeholders=template.placeholders
+    # ).get_rows():
+    #     if row.index == job_row_number:
+    #         notification_id = process_row(row, template, job, job.service)
+    #         current_app.logger.info("Process row {} for job {} created notification_id: {}".format(
+    #             job_row_number, job_id, notification_id))
+
+    pass

app/dao/jobs_dao.py

@@ -8,6 +8,7 @@ from sqlalchemy import (
     asc,
     desc,
     func,
+    and_
 )
 
 from app import db
@@ -187,3 +188,38 @@ def can_letter_job_be_cancelled(job):
         return False, "It’s too late to cancel sending, these letters have already been sent."
 
     return True, None
+
+
+def find_jobs_with_missing_rows():
+    jobs_with_rows_missing = db.session.query(
+        func.count(Notification.id).label('count_notifications'),
+        Job.notification_count,
+        Job.id.label('job_id'),
+        Job.service_id
+    ).filter(
+        Job.job_status == JOB_STATUS_FINISHED
+    ).group_by(
+        Job.notification_count,
+        Job.id.label('job_id'),
+        Job.service_id
+    ).having(
+        func.count(Notification.id) != Job.notification_count
+    )
+
+    return jobs_with_rows_missing.all()
+
+
+def find_missing_row_for_job(job_id, job_size):
+    expected_row_numbers = db.session.query(
+        func.generate_series(0, job_size-1).label('row')
+    ).subquery()
+
+    query = db.session.query(
+        Notification.job_row_number,
+        expected_row_numbers.c.row.label('missing_row')
+    ).outerjoin(
+        Notification, and_(expected_row_numbers.c.row == Notification.job_row_number, Notification.job_id == job_id)
+    ).filter(
+        Notification.job_row_number == None  #noqa
+    )
+    return query.all()