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

Merge branch 'master' into remove-nasty-query-from-dashboard

Browse Source 
Conflicts:
	app/service/rest.py
	tests/app/service/test_rest.py
This commit is contained in:
Martyn Inglis
2017-06-01 13:11:54 +01:00
parent 390935e82c 40aadf2748
commit 7f65aa3eef
64 changed files with 2012 additions and 648 deletions
Download Patch File Download Diff File Expand all files Collapse all files

445
tests/app/service/test_rest.py
View File

@@ -10,7 +10,7 @@ from freezegun import freeze_time
from app.dao.users_dao import save_model_user
from app.dao.services_dao import dao_remove_user_from_service
from app.models import User, Organisation, DVLA_ORG_LAND_REGISTRY, Rate
from app.models import User, Organisation, DVLA_ORG_LAND_REGISTRY, Rate, ServicePermission
from tests import create_authorization_header
from tests.app.db import create_template
from tests.app.conftest import (
@@ -20,28 +20,30 @@ from tests.app.conftest import (
sample_notification_history as create_notification_history,
sample_notification_with_job
)
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST
from app.models import (
Service, ServicePermission,
KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST,
EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INBOUND_SMS_TYPE
)
from tests.app.db import create_user
def test_get_service_list(notify_api, service_factory):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
assert json_resp['data'][2]['name'] == 'three'
def test_get_service_list(client, service_factory):
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
assert json_resp['data'][2]['name'] == 'three'
def test_get_service_list_with_only_active_flag(client, service_factory):
@@ -117,17 +119,15 @@ def test_get_service_list_by_user_should_return_empty_list_if_no_services(client
assert len(json_resp['data']) == 0
def test_get_service_list_should_return_empty_list_if_no_services(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_get_service_list_should_return_empty_list_if_no_services(client):
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_get_service_by_id(client, sample_service):
@@ -147,6 +147,32 @@ def test_get_service_by_id(client, sample_service):
assert json_resp['data']['sms_sender'] == current_app.config['FROM_NUMBER']
def test_get_service_list_has_default_permissions(client, service_factory):
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert all([set(json['permissions']) == set([EMAIL_TYPE, SMS_TYPE]) for json in json_resp['data']])
def test_get_service_by_id_has_default_service_permissions(client, sample_service):
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert set(json_resp['data']['permissions']) == set([EMAIL_TYPE, SMS_TYPE])
def test_get_service_by_id_should_404_if_no_service(notify_api, notify_db):
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -216,6 +242,10 @@ def test_create_service(client, sample_user):
assert json_resp['data']['dvla_organisation'] == '001'
assert json_resp['data']['sms_sender'] == current_app.config['FROM_NUMBER']
service_db = Service.query.get(json_resp['data']['id'])
assert service_db.name == 'created service'
assert service_db.sms_sender == current_app.config['FROM_NUMBER']
auth_header_fetch = create_authorization_header()
resp = client.get(
@@ -410,39 +440,194 @@ def test_update_service(client, notify_db, sample_service):
assert result['data']['dvla_organisation'] == DVLA_ORG_LAND_REGISTRY
def test_update_service_flags(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert json_resp['data']['name'] == sample_service.name
assert json_resp['data']['research_mode'] is False
assert json_resp['data']['can_send_letters'] is False
assert json_resp['data']['can_send_international_sms'] is False
def test_update_service_flags(client, sample_service):
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert json_resp['data']['name'] == sample_service.name
assert json_resp['data']['research_mode'] is False
assert json_resp['data']['can_send_letters'] is False
assert json_resp['data']['can_send_international_sms'] is False
data = {
'research_mode': True,
'can_send_letters': True,
'can_send_international_sms': True,
}
data = {
'research_mode': True,
'can_send_letters': True,
'can_send_international_sms': True,
}
auth_header = create_authorization_header()
auth_header = create_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['research_mode'] is True
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['research_mode'] is True
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
@pytest.fixture(scope='function')
def service_with_no_permissions(notify_db, notify_db_session):
return create_service(notify_db, notify_db_session, permissions=[])
def test_update_service_flags_with_service_without_default_service_permissions(client, service_with_no_permissions):
auth_header = create_authorization_header()
data = {
'can_send_letters': True,
'can_send_international_sms': True,
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
assert set(result['data']['permissions']) == set([LETTER_TYPE, INTERNATIONAL_SMS_TYPE])
def test_update_service_flags_will_remove_service_permissions(client, notify_db, notify_db_session):
auth_header = create_authorization_header()
service = create_service(
notify_db, notify_db_session, permissions=[SMS_TYPE, EMAIL_TYPE, INTERNATIONAL_SMS_TYPE])
assert service.can_send_international_sms is True
data = {
'can_send_international_sms': False
}
resp = client.post(
'/service/{}'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_international_sms'] is False
permissions = ServicePermission.query.filter_by(service_id=service.id).all()
assert set([p.permission for p in permissions]) == set([SMS_TYPE, EMAIL_TYPE])
def test_update_permissions_will_override_permission_flags(client, service_with_no_permissions):
auth_header = create_authorization_header()
data = {
'permissions': [LETTER_TYPE, INTERNATIONAL_SMS_TYPE]
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
assert set(result['data']['permissions']) == set([LETTER_TYPE, INTERNATIONAL_SMS_TYPE])
def test_update_service_permissions_will_add_service_permissions(client, sample_service):
auth_header = create_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert set(result['data']['permissions']) == set([SMS_TYPE, EMAIL_TYPE, LETTER_TYPE])
@pytest.mark.parametrize(
'permission_to_add',
[
(EMAIL_TYPE),
(SMS_TYPE),
(INTERNATIONAL_SMS_TYPE),
(LETTER_TYPE),
(INBOUND_SMS_TYPE),
]
)
def test_add_service_permission_will_add_permission(client, service_with_no_permissions, permission_to_add):
auth_header = create_authorization_header()
data = {
'permissions': [permission_to_add]
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
permissions = ServicePermission.query.filter_by(service_id=service_with_no_permissions.id).all()
assert resp.status_code == 200
assert [p.permission for p in permissions] == [permission_to_add]
def test_update_permissions_with_an_invalid_permission_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
invalid_permission = 'invalid_permission'
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, invalid_permission]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 400
assert result['result'] == 'error'
assert "Invalid Service Permission: '{}'".format(invalid_permission) in result['message']['permissions']
def test_update_permissions_with_duplicate_permissions_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, LETTER_TYPE]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 400
assert result['result'] == 'error'
assert "Duplicate Service Permission: ['{}']".format(LETTER_TYPE) in result['message']['permissions']
def test_update_service_research_mode_throws_validation_error(notify_api, sample_service):
@@ -1609,48 +1794,59 @@ def test_get_monthly_billing_usage_returns_empty_list_if_no_notifications(client
def test_search_for_notification_by_to_field(client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(to_field='jack@gmail.com', normalised_to='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, 'jack@gmail.com'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "jack@gmail.com"),
headers=[create_authorization_header()])
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert len(result["notifications"]) == 1
assert result["notifications"][0]["id"] == str(notification2.id)
assert len(notifications) == 1
assert str(notification2.id) == notifications[0]['id']
def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_match(
client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
client, notify_db, notify_db_session
):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855')
create_notification(to_field='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, '+447700900800'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "+447700900800"),
headers=[create_authorization_header()])
assert response.status_code == 200
assert len(json.loads(response.get_data(as_text=True))["notifications"]) == 0
assert len(notifications) == 0
def test_search_for_notification_by_to_field_return_multiple_matches(
client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session,
to_field=" +44 77009 00855 ")
notification3 = create_sample_notification(notify_db, notify_db_session,
to_field="+44770 0900 855")
notification4 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
def test_search_for_notification_by_to_field_return_multiple_matches(client, notify_db, notify_db_session):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(to_field=' +44 77009 00855 ', normalised_to='447700900855')
notification3 = create_notification(to_field='+44770 0900 855', normalised_to='447700900855')
notification4 = create_notification(to_field='jack@gmail.com', normalised_to='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, '+447700900855'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "+447700900855"),
headers=[create_authorization_header()])
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert len(result["notifications"]) == 3
assert str(notification1.id) in [n["id"] for n in result["notifications"]]
assert str(notification2.id) in [n["id"] for n in result["notifications"]]
assert str(notification3.id) in [n["id"] for n in result["notifications"]]
assert len(notifications) == 3
assert str(notification1.id) in notification_ids
assert str(notification2.id) in notification_ids
assert str(notification3.id) in notification_ids
assert str(notification4.id) not in notification_ids
def test_update_service_calls_send_notification_as_service_becomes_live(notify_db, notify_db_session, client, mocker):
@@ -1808,3 +2004,70 @@ def test_get_yearly_billing_usage_count_returns_from_cache_if_present(client, sa
mock_year.assert_not_called()
mock_query.assert_not_called()
redis_set_mock.assert_not_called()
def test_update_service_works_when_sms_sender_is_null(sample_service, client, mocker):
sample_service.sms_sender = None
data = {'name': 'new name'}
resp = client.post(
'service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[create_authorization_header()],
content_type='application/json'
)
assert resp.status_code == 200
# make sure it wasn't changed to not-null under the hood
assert sample_service.sms_sender is None
def test_search_for_notification_by_to_field_filters_by_status(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
create_notification(status='sending')
response = client.get(
'/service/{}/notifications?to={}&status={}'.format(
notification1.service_id, '+447700900855', 'delivered'
),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
assert response.status_code == 200
assert len(notifications) == 1
assert str(notification1.id) in notification_ids
def test_search_for_notification_by_to_field_filters_by_statuses(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
notification2 = create_notification(status='sending')
response = client.get(
'/service/{}/notifications?to={}&status={}&status={}'.format(
notification1.service_id, '+447700900855', 'delivered', 'sending'
),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
assert response.status_code == 200
assert len(notifications) == 2
assert str(notification1.id) in notification_ids
assert str(notification2.id) in notification_ids