darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-01 15:46:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

separate nightly tasks and other scheduled tasks.

Browse Source 
other tasks is anything that is run on a different frequency than
nightly
This commit is contained in:
Leo Hemsted
2019-01-14 17:22:41 +00:00
parent ef4b2c0564
commit d3d56a3224
6 changed files with 340 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -905,3 +905,21 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
queue='send-email-tasks')
sms_delivery_queue.assert_called_once_with([str(old_sms.id)],
queue="send-sms-tasks")
def test_check_job_status_task_does_not_raise_error(sample_template):
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)
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()

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

@@ -15,7 +15,6 @@ from notifications_utils.columns import Row
from app import (encryption, DATETIME_FORMAT)
from app.celery import provider_tasks
from app.celery import tasks
from app.celery.scheduled_tasks import check_job_status
from app.celery.tasks import (
process_job,
process_row,
@@ -1396,24 +1395,6 @@ def test_send_inbound_sms_to_service_does_not_retries_if_request_returns_404(not
mocked.call_count == 0
def test_check_job_status_task_does_not_raise_error(sample_template):
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)
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()
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'))