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

[WIP] Added endpoints under /job for creating, updating and reading

Browse Source 
notification status.
This commit is contained in:
Adam Shimali
2016-02-09 14:17:42 +00:00
parent 6007cdc24e
commit 17e5e70f6c
5 changed files with 205 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tests/app/dao/test_notification_dao.py
View File

@@ -5,7 +5,7 @@ from app.models import Notification
from app.dao.notifications_dao import (
save_notification,
get_notification,
get_notifications_by_job
get_notifications
)
@@ -39,7 +39,9 @@ def test_save_notification(notify_db, notify_db_session, sample_template, sample
def test_get_notification_for_job(notify_db, notify_db_session, sample_notification):
notifcation_from_db = get_notification(sample_notification.job_id, sample_notification.id)
notifcation_from_db = get_notification(sample_notification.service.id,
sample_notification.job_id,
sample_notification.id)
assert sample_notification == notifcation_from_db
@@ -53,7 +55,7 @@ def test_get_all_notifications_for_job(notify_db, notify_db_session, sample_job)
template=sample_job.template,
job=sample_job)
notifcations_from_db = get_notifications_by_job(sample_job.id)
notifcations_from_db = get_notifications(sample_job.service.id, sample_job.id)
assert len(notifcations_from_db) == 5
@@ -61,10 +63,10 @@ def test_update_notification(notify_db, notify_db_session, sample_notification):
assert sample_notification.status == 'sent'
update_dict = {
'id': sample_notification.id,
'service': sample_notification.service,
'template': sample_notification.template,
'job': sample_notification.job,
'id': str(sample_notification.id),
'service': str(sample_notification.service.id),
'template': sample_notification.template.id,
'job': str(sample_notification.job.id),
'status': 'failed'
}

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

@@ -163,6 +163,130 @@ def test_get_update_job_status(notify_api,
assert resp_json['data']['status'] == 'in progress'
def test_get_notification(notify_api, notify_db, notify_db_session, sample_notification):
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.get_notification_for_job',
service_id=sample_notification.service.id,
job_id=sample_notification.job.id,
notification_id=sample_notification.id)
auth_header = create_authorization_header(service_id=sample_notification.service.id,
path=path,
method='GET')
headers = [('Content-Type', 'application/json'), auth_header]
response = client.get(path, headers=headers)
resp_json = json.loads(response.get_data(as_text=True))
assert str(sample_notification.id) == resp_json['data']['id']
assert str(sample_notification.service.id) == resp_json['data']['service']
assert sample_notification.template.id == resp_json['data']['template']
assert str(sample_notification.job.id) == resp_json['data']['job']
assert sample_notification.status == resp_json['data']['status']
def test_get_notifications(notify_api, notify_db, notify_db_session, sample_job):
from tests.app.conftest import sample_notification
for i in range(0, 5):
sample_notification(notify_db,
notify_db_session,
service=sample_job.service,
template=sample_job.template,
job=sample_job)
service_id = str(sample_job.service.id)
job_id = str(sample_job.id)
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.get_notification_for_job',
service_id=service_id,
job_id=job_id)
auth_header = create_authorization_header(service_id=service_id,
path=path,
method='GET')
headers = [('Content-Type', 'application/json'), auth_header]
response = client.get(path, headers=headers)
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 5
def test_add_notification(notify_api, notify_db, notify_db_session, sample_job):
notificaton_id = uuid.uuid4()
to = '+44709123456'
data = {
'id': str(notificaton_id),
'to': to,
'job': str(sample_job.id),
'service': str(sample_job.service.id),
'template': sample_job.template.id
}
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.create_notification_for_job',
service_id=sample_job.service.id,
job_id=sample_job.id)
auth_header = create_authorization_header(service_id=sample_job.service.id,
path=path,
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
response = client.post(path, headers=headers, data=json.dumps(data))
resp_json = json.loads(response.get_data(as_text=True))
assert data['id'] == resp_json['data']['id']
assert data['to'] == resp_json['data']['to']
assert data['service'] == resp_json['data']['service']
assert data['template'] == resp_json['data']['template']
assert data['job'] == resp_json['data']['job']
assert 'sent' == resp_json['data']['status']
def test_update_notification(notify_api, notify_db, notify_db_session, sample_notification):
assert sample_notification.status == 'sent'
update_data = {
'id': str(sample_notification.id),
'to': sample_notification.to,
'job': str(sample_notification.job.id),
'service': str(sample_notification.service.id),
'template': sample_notification.template.id,
'status': 'failed'
}
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.update_notification_for_job',
service_id=sample_notification.service.id,
job_id=sample_notification.job.id,
notification_id=sample_notification.id)
auth_header = create_authorization_header(service_id=sample_notification.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))
resp_json = json.loads(response.get_data(as_text=True))
assert update_data['id'] == resp_json['data']['id']
assert 'failed' == resp_json['data']['status']
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,