darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-04 10:21:14 -05:00
Code Issues Packages Projects Releases Wiki Activity

Create new task to build dvla file.

Browse Source 
This will transform each notification in a job to a row in a file.
The file is then uploaded to S3.
The files will later be aggregated by the notifications-ftp app to send to dvla.

The method to upload the file to S3 should be pulled into notifications-utils package.
It is the same method used in notifications-admin.
This commit is contained in:
Rebecca Law
2017-03-15 15:26:58 +00:00
parent ea4214c7d5
commit 140179b4b6
6 changed files with 169 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
tests/app/dao/test_jobs_dao.py
View File

@@ -13,7 +13,8 @@ from app.dao.jobs_dao import (
dao_get_future_scheduled_job_by_id_and_service_id,
dao_get_notification_outcomes_for_job,
dao_get_jobs_older_than,
are_all_notifications_created_for_job)
all_notifications_are_created_for_job,
dao_get_all_notifications_for_job)
from app.models import Job
from tests.app.conftest import sample_notification as create_notification
@@ -316,18 +317,28 @@ def test_get_jobs_for_service_doesnt_return_test_messages(notify_db, notify_db_s
assert jobs == [sample_job]
def test_are_all_notifications_created_for_job_returns_true(notify_db, notify_db_session, sample_job):
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=sample_job)
job_is_complete = are_all_notifications_created_for_job(sample_job.id)
def test_all_notifications_are_created_for_job_returns_true(notify_db, notify_db_session):
job = create_job(notify_db=notify_db, notify_db_session=notify_db_session, notification_count=2)
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=job)
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=job)
job_is_complete = all_notifications_are_created_for_job(job.id)
assert job_is_complete
def test_are_all_notifications_created_for_job_returns_false(notify_db, notify_db_session):
def test_all_notifications_are_created_for_job_returns_false(notify_db, notify_db_session):
job = create_job(notify_db=notify_db, notify_db_session=notify_db_session, notification_count=2)
job_is_complete = are_all_notifications_created_for_job(job.id)
job_is_complete = all_notifications_are_created_for_job(job.id)
assert not job_is_complete
def test_are_all_notifications_created_for_job_returns_false_when_job_does_not_exist(notify_db, notify_db_session):
job_is_complete = are_all_notifications_created_for_job(uuid.uuid4())
def test_are_all_notifications_created_for_job_returns_false_when_job_does_not_exist():
job_is_complete = all_notifications_are_created_for_job(uuid.uuid4())
assert not job_is_complete
def test_dao_get_all_notifications_for_job(notify_db, notify_db_session, sample_job):
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=sample_job)
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=sample_job)
create_notification(notify_db=notify_db, notify_db_session=notify_db_session, job=sample_job)
assert len(dao_get_all_notifications_for_job(sample_job.id)) == 3