darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-23 09:29:43 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge with master.

Browse Source
This commit is contained in:
Nicholas Staples
2016-03-01 17:18:46 +00:00
parent 918d40cc9d 9fe08d480c
commit 47c4368007
16 changed files with 520 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
tests/app/conftest.py
View File

@@ -251,7 +251,8 @@ def sample_notification(notify_db,
notify_db_session,
service=None,
template=None,
job=None):
job=None,
to_field=None):
if service is None:
service = sample_service(notify_db, notify_db_session)
if template is None:
@@ -259,11 +260,15 @@ def sample_notification(notify_db,
if job is None:
job = sample_job(notify_db, notify_db_session, service=service, template=template)
notificaton_id = uuid.uuid4()
to = '+44709123456'
notification_id = uuid.uuid4()
if to_field:
to = to_field
else:
to = '+44709123456'
data = {
'id': notificaton_id,
'id': notification_id,
'to': to,
'job': job,
'service': service,

11
tests/app/dao/test_invited_user_dao.py
View File

@@ -68,3 +68,14 @@ def test_get_invited_users_for_service_that_has_no_invites(notify_db, notify_db_
invites = get_invited_users_for_service(sample_service.id)
assert len(invites) == 0
def test_save_invited_user_sets_status_to_cancelled(notify_db, notify_db_session, sample_invited_user):
assert InvitedUser.query.count() == 1
saved = InvitedUser.query.get(sample_invited_user.id)
assert saved.status == 'pending'
saved.status = 'cancelled'
save_invited_user(saved)
assert InvitedUser.query.count() == 1
cancelled_invited_user = InvitedUser.query.get(sample_invited_user.id)
assert cancelled_invited_user.status == 'cancelled'

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

@@ -82,7 +82,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_for_job(sample_job.service.id, sample_job.id)
notifcations_from_db = get_notifications_for_job(sample_job.service.id, sample_job.id).items
assert len(notifcations_from_db) == 5

55
tests/app/invite/test_invite_rest.py
View File

@@ -196,3 +196,58 @@ def test_get_invited_user_by_service_but_unknown_invite_id_returns_404(notify_ap
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 404
def test_update_invited_user_set_status_to_cancelled(notify_api, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {'status': 'cancelled'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header(
path=url,
method='POST',
request_body=json.dumps(data)
)
response = client.post(url,
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
print(json_resp)
assert json_resp['status'] == 'cancelled'
def test_update_invited_user_for_wrong_service_returns_404(notify_api, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {'status': 'cancelled'}
bad_service_id = uuid.uuid4()
url = '/service/{0}/invite/{1}'.format(bad_service_id, sample_invited_user.id)
auth_header = create_authorization_header(
path=url,
method='POST',
request_body=json.dumps(data)
)
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
json_response = json.loads(response.get_data(as_text=True))['message']
assert json_response == 'Invited user not found for service id: {} and invited user id: {}'\
.format(bad_service_id, sample_invited_user.id)
def test_update_invited_user_for_invalid_data_returns_400(notify_api, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {'status': 'garbage'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header(
path=url,
method='POST',
request_body=json.dumps(data)
)
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400

200
tests/app/notifications/test_rest.py
View File

@@ -1,6 +1,7 @@
import uuid
import app.celery.tasks
from tests import create_authorization_header
from tests.app.conftest import sample_notification, sample_job, sample_service
from flask import json
from app.models import Service
from app.dao.templates_dao import dao_get_all_templates_for_service
@@ -47,6 +48,205 @@ def test_get_notifications_empty_result(notify_api, sample_api_key):
assert response.status_code == 404
def test_get_all_notifications(notify_api, sample_notification):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_notification.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert notifications['notifications'][0]['status'] == 'sent'
assert notifications['notifications'][0]['template'] == sample_notification.template.id
assert notifications['notifications'][0]['to'] == '+44709123456'
assert notifications['notifications'][0]['service'] == str(sample_notification.service_id)
assert response.status_code == 200
def test_get_all_notifications_newest_first(notify_api, notify_db, notify_db_session, sample_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
notification_1 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_2 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_3 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
auth_header = create_authorization_header(
service_id=sample_email_template.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications['notifications']) == 3
assert notifications['notifications'][0]['to'] == notification_3.to
assert notifications['notifications'][1]['to'] == notification_2.to
assert notifications['notifications'][2]['to'] == notification_1.to
assert response.status_code == 200
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_1 = sample_service(notify_db, notify_db_session, service_name="1")
service_2 = sample_service(notify_db, notify_db_session, service_name="2")
sample_notification(notify_db, notify_db_session, service=service_2)
notification_1 = sample_notification(notify_db, notify_db_session, service=service_1)
notification_2 = sample_notification(notify_db, notify_db_session, service=service_1)
notification_3 = sample_notification(notify_db, notify_db_session, service=service_1)
auth_header = create_authorization_header(
path='/service/{}/notifications'.format(service_1.id),
method='GET')
response = client.get(
path='/service/{}/notifications'.format(service_1.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_3.to
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][2]['to'] == notification_1.to
assert response.status_code == 200
def test_get_all_notifications_for_job_in_order(notify_api, notify_db, notify_db_session, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
main_job = sample_job(notify_db, notify_db_session, service=sample_service)
another_job = sample_job(notify_db, notify_db_session, service=sample_service)
from time import sleep
notification_1 = sample_notification(notify_db, notify_db_session, job=main_job, to_field="1")
notification_2 = sample_notification(notify_db, notify_db_session, job=main_job, to_field="2")
notification_3 = sample_notification(notify_db, notify_db_session, job=main_job, to_field="3")
sample_notification(notify_db, notify_db_session, job=another_job)
auth_header = create_authorization_header(
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
method='GET')
response = client.get(
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_3.to
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][2]['to'] == notification_1.to
assert response.status_code == 200
def test_should_not_get_notifications_by_service_with_client_credentials(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_api_key.service.id,
path='/service/{}/notifications'.format(sample_api_key.service.id),
method='GET')
response = client.get(
'/service/{}/notifications'.format(sample_api_key.service.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 403
assert resp['result'] == 'error'
assert resp['message'] == 'Forbidden, invalid authentication token provided'
def test_should_not_get_notifications_by_job_and_service_with_client_credentials(notify_api, sample_job):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_job.service.id,
path='/service/{}/job/{}/notifications'.format(sample_job.service.id, sample_job.id),
method='GET')
response = client.get(
'/service/{}/job/{}/notifications'.format(sample_job.service.id, sample_job.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 403
assert resp['result'] == 'error'
assert resp['message'] == 'Forbidden, invalid authentication token provided'
def test_should_reject_invalid_page_param(notify_api, sample_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_email_template.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications?page=invalid',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert notifications['result'] == 'error'
assert notifications['message'] == 'Invalid page'
def test_should_return_pagination_links(notify_api, notify_db, notify_db_session, sample_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
notify_api.config['PAGE_SIZE'] = 1
sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_2 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
sample_notification(notify_db, notify_db_session, sample_email_template.service)
auth_header = create_authorization_header(
service_id=sample_email_template.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications?page=2',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications['notifications']) == 1
assert notifications['links']['last'] == '/notifications?page=3'
assert notifications['links']['prev'] == '/notifications?page=1'
assert notifications['links']['next'] == '/notifications?page=3'
assert notifications['notifications'][0]['to'] == notification_2.to
assert response.status_code == 200
def test_get_all_notifications_returns_empty_list(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_api_key.service.id,
path='/notifications',
method='GET')
response = client.get(
'/notifications',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(notifications['notifications']) == 0
def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:

16
tests/app/user/test_rest.py
View File

@@ -21,7 +21,9 @@ def test_get_user_list(notify_api, notify_db, notify_db_session, sample_service)
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
sample_user = sample_service.users[0]
expected_permissions = ['manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']
expected_permissions = [
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates',
'manage_team', 'view_activity']
fetched = json_resp['data'][0]
assert sample_user.id == fetched['id']
@@ -47,7 +49,9 @@ def test_get_user(notify_api, notify_db, notify_db_session, sample_service):
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = ['manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']
expected_permissions = [
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates',
'manage_team', 'view_activity']
fetched = json_resp['data']
assert sample_user.id == fetched['id']
@@ -180,7 +184,9 @@ def test_put_user(notify_api, notify_db, notify_db_session, sample_service):
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == new_email
expected_permissions = ['manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']
expected_permissions = [
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates',
'manage_team', 'view_activity']
fetched = json_resp['data']
assert sample_user.id == fetched['id']
@@ -272,7 +278,9 @@ def test_get_user_by_email(notify_api, notify_db, notify_db_session, sample_serv
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = ['manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']
expected_permissions = [
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates',
'manage_team', 'view_activity']
fetched = json_resp['data']
assert sample_user.id == fetched['id']