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

Endpoint and dao method for updating job status.

Browse Source
This commit is contained in:
Adam Shimali
2016-02-04 20:55:09 +00:00
parent c395069bb9
commit bec4bbe04e
4 changed files with 84 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,4 +1,5 @@
import uuid
import json
from app.dao.jobs_dao import (
save_job,
@@ -79,3 +80,23 @@ def test_get_all_jobs(notify_db, notify_db_session, sample_template):
sample_template)
jobs_from_db = _get_jobs()
assert len(jobs_from_db) == 5
def test_update_job(notify_db, notify_db_session, sample_job):
assert sample_job.status == 'pending'
update_dict = {
'id': sample_job.id,
'service': sample_job.service.id,
'template': sample_job.template.id,
'bucket_name': sample_job.bucket_name,
'file_name': sample_job.file_name,
'original_file_name': sample_job.original_file_name,
'status': 'in progress'
}
save_job(sample_job, update_dict=update_dict)
job_from_db = Job.query.get(sample_job.id)
assert job_from_db.status == 'in progress'

38
tests/app/job/test_job_rest.py
View File

@@ -125,6 +125,44 @@ def test_create_job(notify_api, notify_db, notify_db_session, sample_template):
assert expected_message['bucket_name'] == bucket_name
def test_get_update_job_status(notify_api,
notify_db,
notify_db_session,
sample_job):
assert sample_job.status == 'pending'
job_id = str(sample_job.id)
service_id = str(sample_job.service.id)
update_data = {
'id': job_id,
'service': service_id,
'template': sample_job.template.id,
'bucket_name': sample_job.bucket_name,
'file_name': sample_job.file_name,
'original_file_name': sample_job.original_file_name,
'status': 'in progress'
}
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.update_job', service_id=service_id, job_id=job_id)
auth_header = create_authorization_header(service_id=service_id,
path=path,
method='PUT',
request_body=json.dumps(update_data))
headers = [('Content-Type', 'application/json'), auth_header]
response = client.put(path, headers=headers, data=json.dumps(update_data))
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['data']['status'] == 'in progress'
def _setup_jobs(notify_db, notify_db_session, template, number_of_jobs=5):
for i in range(number_of_jobs):
create_job(notify_db, notify_db_session, service=template.service,