mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-24 00:06:16 -04:00
b3884e2d6c
- brings boto S3 into new AWS folder - CSV processing utils method Rejigs the jobs rest endpoint - removes some now unused endpoints, Calls to the task with the job, job processing in task, delegating SMS calls to the sms task
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
import uuid
|
|
|
|
from app.dao.jobs_dao import (
|
|
dao_get_job_by_service_id_and_job_id,
|
|
dao_create_job,
|
|
dao_update_job,
|
|
dao_get_jobs_by_service_id
|
|
)
|
|
|
|
from app.models import Job
|
|
|
|
|
|
def test_create_job(sample_template):
|
|
assert Job.query.count() == 0
|
|
|
|
job_id = uuid.uuid4()
|
|
bucket_name = 'service-{}-notify'.format(sample_template.service.id)
|
|
file_name = '{}.csv'.format(job_id)
|
|
data = {
|
|
'id': job_id,
|
|
'service_id': sample_template.service.id,
|
|
'template_id': sample_template.id,
|
|
'bucket_name': bucket_name,
|
|
'file_name': file_name,
|
|
'original_file_name': 'some.csv',
|
|
'notification_count': 1
|
|
}
|
|
|
|
job = Job(**data)
|
|
dao_create_job(job)
|
|
|
|
assert Job.query.count() == 1
|
|
job_from_db = Job.query.get(job_id)
|
|
assert job == job_from_db
|
|
|
|
|
|
def test_get_job_by_id(sample_job):
|
|
job_from_db = dao_get_job_by_service_id_and_job_id(sample_job.service.id, sample_job.id)
|
|
assert sample_job == job_from_db
|
|
|
|
|
|
def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
|
|
from tests.app.conftest import sample_job as create_job
|
|
from tests.app.conftest import sample_service as create_service
|
|
from tests.app.conftest import sample_template as create_template
|
|
from tests.app.conftest import sample_user as create_user
|
|
|
|
one_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template)
|
|
|
|
other_user = create_user(notify_db, notify_db_session, email="test@digital.cabinet-office.gov.uk")
|
|
other_service = create_service(notify_db, notify_db_session, user=other_user, service_name="other service")
|
|
other_template = create_template(notify_db, notify_db_session, service=other_service)
|
|
other_job = create_job(notify_db, notify_db_session, service=other_service, template=other_template)
|
|
|
|
one_job_from_db = dao_get_jobs_by_service_id(one_job.service_id)
|
|
other_job_from_db = dao_get_jobs_by_service_id(other_job.service_id)
|
|
|
|
assert len(one_job_from_db) == 1
|
|
assert one_job == one_job_from_db[0]
|
|
|
|
assert len(other_job_from_db) == 1
|
|
assert other_job == other_job_from_db[0]
|
|
|
|
assert one_job_from_db != other_job_from_db
|
|
|
|
|
|
def test_update_job(sample_job):
|
|
assert sample_job.status == 'pending'
|
|
|
|
sample_job.status = 'in progress'
|
|
|
|
dao_update_job(sample_job)
|
|
|
|
job_from_db = Job.query.get(sample_job.id)
|
|
|
|
assert job_from_db.status == 'in progress'
|