If a job was created 7 days ago but scheduled to send 2 days later, the csv file will be deleted from S3 two days early.

For example if a file is uploaded on Aug 1, but scheduled for Aug 3, the csv file will be deleted 2 days before the notifications.
This will cause an error for on the jobs pages if the download report link is clicked.

app/dao/jobs_dao.py

@@ -134,7 +134,7 @@ def dao_get_jobs_older_than_data_retention(notification_types):
         end_date = today - timedelta(days=f.days_of_retention)
 
         jobs.extend(Job.query.join(Template).filter(
-            Job.created_at < end_date,
+            func.coalesce(Job.scheduled_for, Job.created_at) < end_date,
             Job.archived == False,  # noqa
             Template.template_type == f.notification_type,
             Job.service_id == f.service_id
@@ -146,7 +146,7 @@ def dao_get_jobs_older_than_data_retention(notification_types):
             x.service_id for x in flexible_data_retention if x.notification_type == notification_type
         ]
         jobs.extend(Job.query.join(Template).filter(
-            Job.created_at < end_date,
+            func.coalesce(Job.scheduled_for, Job.created_at) < end_date,
             Job.archived == False,  # noqa
             Template.template_type == notification_type,
             Job.service_id.notin_(services_with_data_retention)

pytest.ini

@@ -6,4 +6,4 @@ env =
     LOADTESTING_API_KEY=loadtesting
     FIRETEXT_API_KEY=Firetext
     NOTIFICATION_QUEUE_PREFIX=testing
-addopts = -n 4 -v -p no:warnings
+addopts = -v -p no:warnings

tests/app/dao/test_jobs_dao.py

@@ -297,6 +297,24 @@ def test_should_get_jobs_seven_days_old_filters_type(sample_service):
     assert job_to_remain.id not in [job.id for job in jobs]
 
 
+@freeze_time('2016-10-31 10:00:00')
+def test_should_get_jobs_seven_days_old_by_scheduled_for_date(sample_service):
+    six_days_ago = datetime.utcnow() - timedelta(days=6)
+    eight_days_ago = datetime.utcnow() - timedelta(days=8)
+    letter_template = create_template(sample_service, template_type=LETTER_TYPE)
+
+    create_job(letter_template, created_at=eight_days_ago)
+    create_job(letter_template, created_at=eight_days_ago, scheduled_for=eight_days_ago)
+    job_to_remain = create_job(letter_template, created_at=eight_days_ago, scheduled_for=six_days_ago)
+
+    jobs = dao_get_jobs_older_than_data_retention(
+        notification_types=[LETTER_TYPE]
+    )
+
+    assert len(jobs) == 2
+    assert job_to_remain.id not in [job.id for job in jobs]
+
+
 def assert_job_stat(job, result, sent, delivered, failed):
     assert result.job_id == job.id
     assert result.original_file_name == job.original_file_name