darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-19 05:58:53 -04:00
Code Issues Packages Projects Releases Wiki Activity

Split generating authorization headers by type

Browse Source 
In response to [1].

[1]: https://github.com/alphagov/notifications-api/pull/3300#discussion_r681653248
This commit is contained in:
Ben Thorner
2021-08-04 15:12:09 +01:00
parent 5a1636e41f
commit 0312e2a528
33 changed files with 526 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
tests/app/service/test_api_key_endpoints.py
View File

@@ -4,7 +4,7 @@ from flask import url_for
from app.dao.api_key_dao import expire_api_key
from app.models import KEY_TYPE_NORMAL, ApiKey
from tests import create_authorization_header
from tests import create_admin_authorization_header
from tests.app.db import create_api_key, create_service, create_user
@@ -16,7 +16,7 @@ def test_api_key_should_create_new_api_key_for_service(notify_api, sample_servic
'created_by': str(sample_service.created_by.id),
'key_type': KEY_TYPE_NORMAL
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post(url_for('service.create_api_key', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -32,7 +32,7 @@ def test_api_key_should_return_error_when_service_does_not_exist(notify_api, sam
with notify_api.test_client() as client:
import uuid
missing_service_id = uuid.uuid4()
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post(url_for('service.create_api_key', service_id=missing_service_id),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
@@ -44,7 +44,7 @@ def test_create_api_key_without_key_type_rejects(notify_api, sample_service):
'name': 'some secret name',
'created_by': str(sample_service.created_by.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post(url_for('service.create_api_key', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -58,7 +58,7 @@ def test_revoke_should_expire_api_key_for_service(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert ApiKey.query.count() == 1
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post(url_for('service.revoke_api_key',
service_id=sample_api_key.service_id,
api_key_id=sample_api_key.id),
@@ -77,7 +77,7 @@ def test_api_key_should_create_multiple_new_api_key_for_service(notify_api, samp
'created_by': str(sample_service.created_by.id),
'key_type': KEY_TYPE_NORMAL
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post(url_for('service.create_api_key', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -85,7 +85,7 @@ def test_api_key_should_create_multiple_new_api_key_for_service(notify_api, samp
assert ApiKey.query.count() == 1
data['name'] = 'another secret name'
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response2 = client.post(url_for('service.create_api_key', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -110,7 +110,7 @@ def test_get_api_keys_should_return_all_keys_for_service(notify_api, sample_api_
assert ApiKey.query.count() == 4
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(url_for('service.get_api_keys',
service_id=sample_api_key.service_id),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -122,7 +122,7 @@ def test_get_api_keys_should_return_all_keys_for_service(notify_api, sample_api_
def test_get_api_keys_should_return_one_key_for_service(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(url_for('service.get_api_keys',
service_id=sample_api_key.service_id,
key_id=sample_api_key.id),

14
tests/app/service/test_archived_service.py
View File

@@ -9,24 +9,24 @@ from app.dao.api_key_dao import expire_api_key
from app.dao.services_dao import dao_archive_service, dao_fetch_service_by_id
from app.dao.templates_dao import dao_update_template
from app.models import Service
from tests import create_authorization_header, unwrap_function
from tests import create_admin_authorization_header, unwrap_function
from tests.app.db import create_api_key, create_template
def test_archive_only_allows_post(client, notify_db_session):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get('/service/{}/archive'.format(uuid.uuid4()), headers=[auth_header])
assert response.status_code == 405
def test_archive_service_errors_with_bad_service_id(client, notify_db_session):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post('/service/{}/archive'.format(uuid.uuid4()), headers=[auth_header])
assert response.status_code == 404
def test_deactivating_inactive_service_does_nothing(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
sample_service.active = False
response = client.post('/service/{}/archive'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204
@@ -42,7 +42,7 @@ def archived_service(client, notify_db, sample_service):
notify_db.session.commit()
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post('/service/{}/archive'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204
assert response.data == b''
@@ -51,7 +51,7 @@ def archived_service(client, notify_db, sample_service):
@freeze_time('2018-07-07 12:00:00')
def test_deactivating_service_changes_name_and_email(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
client.post('/service/{}/archive'.format(sample_service.id), headers=[auth_header])
archived_service = dao_fetch_service_by_id(sample_service.id)
@@ -98,7 +98,7 @@ def archived_service_with_deleted_stuff(client, sample_service):
dao_update_template(template)
with freeze_time('2002-02-02'):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post('/service/{}/archive'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204

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

@@ -45,7 +45,7 @@ from app.models import (
ServiceSmsSender,
User,
)
from tests import create_authorization_header
from tests import create_admin_authorization_header
from tests.app.db import (
create_annual_billing,
create_api_key,
@@ -76,7 +76,7 @@ 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()
auth_header = create_admin_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
@@ -95,7 +95,7 @@ def test_get_service_list_with_only_active_flag(client, service_factory):
inactive.active = False
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
'/service?only_active=True',
headers=[auth_header]
@@ -359,7 +359,7 @@ def test_get_service_by_id_should_404_if_no_service(admin_request, notify_db_ses
def test_get_service_by_id_and_user(client, sample_service, sample_user):
sample_service.reply_to_email = 'something@service.com'
create_reply_to_email(service=sample_service, email_address='new@service.com')
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}?user_id={}'.format(sample_service.id, sample_user.id),
headers=[auth_header]
@@ -374,7 +374,7 @@ def test_get_service_by_id_should_404_if_no_service_for_user(notify_api, sample_
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_id = str(uuid.uuid4())
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}?user_id={}'.format(service_id, sample_user.id),
headers=[auth_header]
@@ -570,7 +570,7 @@ def test_should_not_create_service_with_missing_user_id_field(notify_api, fake_u
'active': False,
'created_by': str(fake_uuid)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -593,7 +593,7 @@ def test_should_error_if_created_by_missing(notify_api, sample_user):
'active': False,
'user_id': str(sample_user.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -620,7 +620,7 @@ def test_should_not_create_service_with_missing_if_user_id_is_not_in_database(no
'active': False,
'created_by': str(fake_uuid)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -638,7 +638,7 @@ def test_should_not_create_service_if_missing_data(notify_api, sample_user):
data = {
'user_id': str(sample_user.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -666,7 +666,7 @@ def test_should_not_create_service_with_duplicate_name(notify_api,
'email_from': 'sample.service2',
'created_by': str(sample_user.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -693,7 +693,7 @@ def test_create_service_should_throw_duplicate_key_constraint_for_existing_email
'email_from': 'first.service',
'created_by': str(sample_user.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -719,7 +719,7 @@ def test_update_service(client, notify_db, sample_service):
'organisation_type': 'school_or_college',
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -742,7 +742,7 @@ def test_cant_update_service_org_type_to_random_value(client, sample_service):
'organisation_type': 'foo',
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -758,7 +758,7 @@ def test_update_service_letter_branding(client, notify_db, sample_service):
'letter_branding': str(letter_branding.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -777,7 +777,7 @@ def test_update_service_remove_letter_branding(client, notify_db, sample_service
'letter_branding': str(letter_branding.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
client.post(
'/service/{}'.format(sample_service.id),
@@ -828,7 +828,7 @@ def test_update_service_change_email_branding(admin_request, notify_db, sample_s
def test_update_service_flags(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
@@ -843,7 +843,7 @@ def test_update_service_flags(client, sample_service):
'permissions': [LETTER_TYPE, INTERNATIONAL_SMS_TYPE]
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -915,7 +915,7 @@ def service_with_no_permissions(notify_db, notify_db_session):
def test_update_service_flags_with_service_without_default_service_permissions(client, service_with_no_permissions):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
data = {
'permissions': [LETTER_TYPE, INTERNATIONAL_SMS_TYPE],
}
@@ -932,7 +932,7 @@ def test_update_service_flags_with_service_without_default_service_permissions(c
def test_update_service_flags_will_remove_service_permissions(client, notify_db, notify_db_session):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
service = create_service(service_permissions=[SMS_TYPE, EMAIL_TYPE, INTERNATIONAL_SMS_TYPE])
@@ -957,7 +957,7 @@ def test_update_service_flags_will_remove_service_permissions(client, notify_db,
def test_update_permissions_will_override_permission_flags(client, service_with_no_permissions):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
data = {
'permissions': [LETTER_TYPE, INTERNATIONAL_SMS_TYPE]
@@ -975,7 +975,7 @@ def test_update_permissions_will_override_permission_flags(client, service_with_
def test_update_service_permissions_will_add_service_permissions(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE]
@@ -1005,7 +1005,7 @@ def test_update_service_permissions_will_add_service_permissions(client, sample_
]
)
def test_add_service_permission_will_add_permission(client, service_with_no_permissions, permission_to_add):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
data = {
'permissions': [permission_to_add]
@@ -1024,7 +1024,7 @@ def test_add_service_permission_will_add_permission(client, service_with_no_perm
def test_update_permissions_with_an_invalid_permission_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
invalid_permission = 'invalid_permission'
data = {
@@ -1044,7 +1044,7 @@ def test_update_permissions_with_an_invalid_permission_will_raise_error(client,
def test_update_permissions_with_duplicate_permissions_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, LETTER_TYPE]
@@ -1065,7 +1065,7 @@ def test_update_permissions_with_duplicate_permissions_will_raise_error(client,
def test_update_service_research_mode_throws_validation_error(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
@@ -1079,7 +1079,7 @@ def test_update_service_research_mode_throws_validation_error(notify_api, sample
'research_mode': "dedede"
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -1108,7 +1108,7 @@ def test_should_not_update_service_with_duplicate_name(notify_api,
'created_by': str(service.created_by.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -1140,7 +1140,7 @@ def test_should_not_update_service_with_duplicate_email_from(notify_api,
'created_by': str(service.created_by.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
@@ -1165,7 +1165,7 @@ def test_update_service_should_404_if_id_is_invalid(notify_api):
missing_service_id = uuid.uuid4()
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}'.format(missing_service_id),
@@ -1179,7 +1179,7 @@ def test_get_users_by_service(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
user_on_service = sample_service.users[0]
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}/users'.format(sample_service.id),
@@ -1199,7 +1199,7 @@ def test_get_users_for_service_returns_empty_list_if_no_users_associated_with_se
with notify_api.test_request_context():
with notify_api.test_client() as client:
dao_remove_user_from_service(sample_service, sample_service.users[0])
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
'/service/{}/users'.format(sample_service.id),
@@ -1214,7 +1214,7 @@ def test_get_users_for_service_returns_404_when_service_does_not_exist(notify_ap
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_id = uuid.uuid4()
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
'/service/{}/users'.format(service_id),
@@ -1242,7 +1242,7 @@ def test_default_permissions_are_added_for_user_service(notify_api,
'email_from': 'created.service',
'created_by': str(sample_user.id)
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
'/service',
@@ -1254,14 +1254,14 @@ def test_default_permissions_are_added_for_user_service(notify_api,
assert json_resp['data']['name'] == 'created service'
assert json_resp['data']['email_from'] == 'created.service'
auth_header_fetch = create_authorization_header()
auth_header_fetch = create_admin_authorization_header()
resp = client.get(
'/service/{}?user_id={}'.format(json_resp['data']['id'], sample_user.id),
headers=[auth_header_fetch]
)
assert resp.status_code == 200
header = create_authorization_header()
header = create_admin_authorization_header()
response = client.get(
url_for('user.get_user', user_id=sample_user.id),
headers=[header])
@@ -1283,7 +1283,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
with notify_api.test_client() as client:
# check which users part of service
user_already_in_service = sample_service.users[0]
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}/users'.format(sample_service.id),
@@ -1319,7 +1319,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
"folder_permissions": []
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, user_to_add.id),
@@ -1330,7 +1330,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
assert resp.status_code == 201
# check new user added to service
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
@@ -1340,7 +1340,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
json_resp = resp.json
# check user has all permissions
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(url_for('user.get_user', user_id=user_to_add.id),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1377,7 +1377,7 @@ def test_add_existing_user_to_another_service_with_send_permissions(notify_api,
"folder_permissions": []
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, user_to_add.id),
@@ -1388,7 +1388,7 @@ def test_add_existing_user_to_another_service_with_send_permissions(notify_api,
assert resp.status_code == 201
# check user has send permissions
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(url_for('user.get_user', user_id=user_to_add.id),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1424,7 +1424,7 @@ def test_add_existing_user_to_another_service_with_manage_permissions(notify_api
]
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, user_to_add.id),
@@ -1435,7 +1435,7 @@ def test_add_existing_user_to_another_service_with_manage_permissions(notify_api
assert resp.status_code == 201
# check user has send permissions
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(url_for('user.get_user', user_id=user_to_add.id),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1471,7 +1471,7 @@ def test_add_existing_user_to_another_service_with_folder_permissions(notify_api
"folder_permissions": [str(folder_1.id), str(folder_2.id)]
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, user_to_add.id),
@@ -1506,7 +1506,7 @@ def test_add_existing_user_to_another_service_with_manage_api_keys(notify_api,
data = {"permissions": [{"permission": "manage_api_keys"}]}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, user_to_add.id),
@@ -1517,7 +1517,7 @@ def test_add_existing_user_to_another_service_with_manage_api_keys(notify_api,
assert resp.status_code == 201
# check user has send permissions
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.get(url_for('user.get_user', user_id=user_to_add.id),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1546,7 +1546,7 @@ def test_add_existing_user_to_non_existing_service_returns404(notify_api,
incorrect_id = uuid.uuid4()
data = {'permissions': ['send_messages', 'manage_service', 'manage_api_keys']}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(incorrect_id, user_to_add.id),
@@ -1568,7 +1568,7 @@ def test_add_existing_user_of_service_to_service_returns400(notify_api, notify_d
existing_user_id = sample_service.users[0].id
data = {'permissions': ['send_messages', 'manage_service', 'manage_api_keys']}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, existing_user_id),
@@ -1590,7 +1590,7 @@ def test_add_unknown_user_to_service_returns404(notify_api, notify_db, notify_db
incorrect_id = 9876
data = {'permissions': ['send_messages', 'manage_service', 'manage_api_keys']}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'/service/{}/users/{}'.format(sample_service.id, incorrect_id),
@@ -1623,7 +1623,7 @@ def test_remove_user_from_service(
'service.remove_user_from_service',
service_id=str(service.id),
user_id=str(second_user.id))
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.delete(
endpoint,
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1638,7 +1638,7 @@ def test_remove_non_existant_user_from_service(
'service.remove_user_from_service',
service_id=str(sample_user_service_permission.service.id),
user_id=str(second_user.id))
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.delete(
endpoint,
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1655,7 +1655,7 @@ def test_cannot_remove_only_user_from_service(notify_api,
'service.remove_user_from_service',
service_id=str(sample_user_service_permission.service.id),
user_id=str(sample_user_service_permission.user.id))
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.delete(
endpoint,
headers=[('Content-Type', 'application/json'), auth_header])
@@ -1669,7 +1669,7 @@ def test_cannot_remove_only_user_from_service(notify_api,
def test_get_service_and_api_key_history(notify_api, sample_service, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
path='/service/{}/history'.format(sample_service.id),
headers=[auth_header]
@@ -1696,7 +1696,7 @@ def test_get_all_notifications_for_service_in_order(notify_api, notify_db_sessio
notification_2 = create_notification(service_1_template)
notification_3 = create_notification(service_1_template)
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
path='/service/{}/notifications'.format(service_1.id),
@@ -1712,7 +1712,7 @@ def test_get_all_notifications_for_service_in_order(notify_api, notify_db_sessio
def test_get_all_notifications_for_service_formatted_for_csv(client, sample_template):
notification = create_notification(template=sample_template)
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
path='/service/{}/notifications?format_for_csv=True'.format(sample_template.service_id),
@@ -1732,7 +1732,7 @@ def test_get_notification_for_service_without_uuid(client, notify_db, notify_db_
service_1 = create_service(service_name="1", email_from='1')
response = client.get(
path='/service/{}/notifications/{}'.format(service_1.id, 'foo'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert response.status_code == 404
@@ -1755,7 +1755,7 @@ def test_get_notification_for_service(client, notify_db_session):
for notification in service_1_notifications:
response = client.get(
path='/service/{}/notifications/{}'.format(service_1.id, notification.id),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
resp = json.loads(response.get_data(as_text=True))
assert str(resp['id']) == str(notification.id)
@@ -1763,7 +1763,7 @@ def test_get_notification_for_service(client, notify_db_session):
service_2_response = client.get(
path='/service/{}/notifications/{}'.format(service_2.id, notification.id),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert service_2_response.status_code == 404
service_2_response = json.loads(service_2_response.get_data(as_text=True))
@@ -1824,7 +1824,7 @@ def test_get_all_notifications_for_service_including_ones_made_by_jobs(
# notification from_test_api_key
create_notification(sample_template, key_type=KEY_TYPE_TEST)
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get(
path='/service/{}/notifications?include_from_test_key={}'.format(
@@ -1903,7 +1903,7 @@ def test_prefixing_messages_based_on_prefix_sms(
'service.get_service_by_id',
service_id=service.id
),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
service = json.loads(result.get_data(as_text=True))['data']
assert service['prefix_sms'] == should_prefix
@@ -1953,7 +1953,7 @@ def test_get_detailed_service(sample_template, notify_api, sample_service, today
create_notification(template=sample_template, status='created')
resp = client.get(
'/service/{}?detailed=True&today_only={}'.format(sample_service.id, today_only),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert resp.status_code == 200
@@ -1972,7 +1972,7 @@ def test_get_services_with_detailed_flag(client, sample_template):
]
resp = client.get(
'/service?detailed=True',
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert resp.status_code == 200
@@ -1997,7 +1997,7 @@ def test_get_services_with_detailed_flag_excluding_from_test_key(notify_api, sam
with notify_api.test_request_context(), notify_api.test_client() as client:
resp = client.get(
'/service?detailed=True&include_from_test_key=False',
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert resp.status_code == 200
@@ -2014,7 +2014,7 @@ def test_get_services_with_detailed_flag_accepts_date_range(client, mocker):
mock_get_detailed_services = mocker.patch('app.service.rest.get_detailed_services', return_value={})
resp = client.get(
url_for('service.get_services', detailed=True, start_date='2001-01-01', end_date='2002-02-02'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
mock_get_detailed_services.assert_called_once_with(
@@ -2031,7 +2031,7 @@ def test_get_services_with_detailed_flag_defaults_to_today(client, mocker):
mock_get_detailed_services = mocker.patch('app.service.rest.get_detailed_services', return_value={})
resp = client.get(
url_for('service.get_services', detailed=True),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
mock_get_detailed_services.assert_called_once_with(
@@ -2165,7 +2165,7 @@ def test_search_for_notification_by_to_field(client, sample_template, sample_ema
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(notification1.service_id, 'jack@gmail.com', 'email'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
@@ -2182,7 +2182,7 @@ def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_ma
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(notification1.service_id, '+447700900800', 'sms'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
@@ -2199,7 +2199,7 @@ def test_search_for_notification_by_to_field_return_multiple_matches(client, sam
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(notification1.service_id, '+447700900855', 'sms'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
@@ -2221,7 +2221,7 @@ def test_search_for_notification_by_to_field_returns_next_link_if_more_than_50(
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(sample_template.service_id, '+447700900855', 'sms'),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
assert response.status_code == 200
response_json = json.loads(response.get_data(as_text=True))
@@ -2246,7 +2246,7 @@ def test_search_for_notification_by_to_field_for_letter(
'/service/{}/notifications?to={}&template_type={}'.format(
sample_letter_template.service_id, 'A. Name', 'letter',
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
@@ -2264,7 +2264,7 @@ def test_update_service_calls_send_notification_as_service_becomes_live(notify_d
"restricted": False
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'service/{}'.format(restricted_service.id),
data=json.dumps(data),
@@ -2291,7 +2291,7 @@ def test_update_service_does_not_call_send_notification_for_live_service(sample_
"restricted": True
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'service/{}'.format(sample_service.id),
data=json.dumps(data),
@@ -2310,7 +2310,7 @@ def test_update_service_does_not_call_send_notification_when_restricted_not_chan
"name": 'Name of service'
}
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
resp = client.post(
'service/{}'.format(sample_service.id),
data=json.dumps(data),
@@ -2331,7 +2331,7 @@ def test_search_for_notification_by_to_field_filters_by_status(client, sample_te
'/service/{}/notifications?to={}&status={}&template_type={}'.format(
notification1.service_id, '+447700900855', 'delivered', 'sms'
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
@@ -2357,7 +2357,7 @@ def test_search_for_notification_by_to_field_filters_by_statuses(client, sample_
'/service/{}/notifications?to={}&status={}&status={}&template_type={}'.format(
notification1.service_id, '+447700900855', 'delivered', 'sending', 'sms'
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
@@ -2383,7 +2383,7 @@ def test_search_for_notification_by_to_field_returns_content(
'/service/{}/notifications?to={}&template_type={}'.format(
sample_template_with_placeholders.service_id, '+447700900855', 'sms'
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
assert response.status_code == 200
@@ -2430,7 +2430,7 @@ def test_create_pdf_letter(mocker, sample_service_full_permissions, client, fake
response = client.post(
url_for('service.create_pdf_letter', service_id=sample_service_full_permissions.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
json_resp = json.loads(response.get_data(as_text=True))
@@ -2464,7 +2464,7 @@ def test_create_pdf_letter_validates_against_json_schema(
response = client.post(
url_for('service.create_pdf_letter', service_id=sample_service_full_permissions.id),
data=json.dumps(post_data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
json_resp = json.loads(response.get_data(as_text=True))
@@ -2559,7 +2559,7 @@ def test_search_for_notification_by_to_field_returns_personlisation(
'/service/{}/notifications?to={}&template_type={}'.format(
sample_template_with_placeholders.service_id, '+447700900855', 'sms'
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
@@ -2582,7 +2582,7 @@ def test_search_for_notification_by_to_field_returns_notifications_by_type(
sms_notification.service_id, '0770', 'sms'
),
headers=[create_authorization_header()]
headers=[create_admin_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
@@ -2686,7 +2686,7 @@ def test_is_service_name_unique_returns_400_when_name_does_not_exist(admin_reque
def test_get_email_reply_to_addresses_when_there_are_no_reply_to_email_addresses(client, sample_service):
response = client.get('/service/{}/email-reply-to'.format(sample_service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
assert json.loads(response.get_data(as_text=True)) == []
assert response.status_code == 200
@@ -2697,7 +2697,7 @@ def test_get_email_reply_to_addresses_with_one_email_address(client, notify_db,
create_reply_to_email(service, 'test@mail.com')
response = client.get('/service/{}/email-reply-to'.format(service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response) == 1
@@ -2714,7 +2714,7 @@ def test_get_email_reply_to_addresses_with_multiple_email_addresses(client, noti
reply_to_b = create_reply_to_email(service, 'test_b@mail.com', False)
response = client.get('/service/{}/email-reply-to'.format(service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response) == 2
@@ -2928,7 +2928,7 @@ def test_get_email_reply_to_address(client, notify_db, notify_db_session):
reply_to = create_reply_to_email(service, 'test_a@mail.com')
response = client.get('/service/{}/email-reply-to/{}'.format(service.id, reply_to.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == reply_to.serialize()
@@ -2936,7 +2936,7 @@ def test_get_email_reply_to_address(client, notify_db, notify_db_session):
def test_get_letter_contacts_when_there_are_no_letter_contacts(client, sample_service):
response = client.get('/service/{}/letter-contact'.format(sample_service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
assert json.loads(response.get_data(as_text=True)) == []
assert response.status_code == 200
@@ -2947,7 +2947,7 @@ def test_get_letter_contacts_with_one_letter_contact(client, notify_db, notify_d
create_letter_contact(service, 'Aberdeen, AB23 1XH')
response = client.get('/service/{}/letter-contact'.format(service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response) == 1
@@ -2964,7 +2964,7 @@ def test_get_letter_contacts_with_multiple_letter_contacts(client, notify_db, no
letter_contact_b = create_letter_contact(service, 'London, E1 8QS', False)
response = client.get('/service/{}/letter-contact'.format(service.id),
headers=[create_authorization_header()])
headers=[create_admin_authorization_header()])
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response) == 2
@@ -2990,7 +2990,7 @@ def test_get_letter_contact_by_id(client, notify_db, notify_db_session):
letter_contact = create_letter_contact(service, 'London, E1 8QS')
response = client.get('/service/{}/letter-contact/{}'.format(service.id, letter_contact.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == letter_contact.serialize()
@@ -3000,7 +3000,7 @@ def test_get_letter_contact_return_404_when_invalid_contact_id(client, notify_db
service = create_service()
response = client.get('/service/{}/letter-contact/{}'.format(service.id, '93d59f88-4aa1-453c-9900-f61e2fc8a2de'),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 404
@@ -3009,7 +3009,7 @@ def test_add_service_contact_block(client, sample_service):
data = json.dumps({"contact_block": "London, E1 8QS", "is_default": True})
response = client.post('/service/{}/letter-contact'.format(sample_service.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
@@ -3022,12 +3022,12 @@ def test_add_service_letter_contact_can_add_multiple_addresses(client, sample_se
first = json.dumps({"contact_block": "London, E1 8QS", "is_default": True})
client.post('/service/{}/letter-contact'.format(sample_service.id),
data=first,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
second = json.dumps({"contact_block": "Aberdeen, AB23 1XH", "is_default": True})
response = client.post('/service/{}/letter-contact'.format(sample_service.id),
data=second,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
results = ServiceLetterContact.query.all()
@@ -3042,14 +3042,14 @@ def test_add_service_letter_contact_block_fine_if_no_default(client, sample_serv
data = json.dumps({"contact_block": "London, E1 8QS", "is_default": False})
response = client.post('/service/{}/letter-contact'.format(sample_service.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 201
def test_add_service_letter_contact_block_404s_when_invalid_service_id(client, notify_db, notify_db_session):
response = client.post('/service/{}/letter-contact'.format(uuid.uuid4()),
data={},
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 404
result = json.loads(response.get_data(as_text=True))
@@ -3062,7 +3062,7 @@ def test_update_service_letter_contact(client, sample_service):
data = json.dumps({"contact_block": "London, E1 8QS", "is_default": True})
response = client.post('/service/{}/letter-contact/{}'.format(sample_service.id, original_letter_contact.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
@@ -3076,14 +3076,14 @@ def test_update_service_letter_contact_returns_200_when_no_default(client, sampl
data = json.dumps({"contact_block": "London, E1 8QS", "is_default": False})
response = client.post('/service/{}/letter-contact/{}'.format(sample_service.id, original_reply_to.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 200
def test_update_service_letter_contact_returns_404_when_invalid_service_id(client, notify_db, notify_db_session):
response = client.post('/service/{}/letter-contact/{}'.format(uuid.uuid4(), uuid.uuid4()),
data={},
headers=[('Content-Type', 'application/json'), create_authorization_header()])
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()])
assert response.status_code == 404
result = json.loads(response.get_data(as_text=True))
@@ -3128,7 +3128,7 @@ def test_add_service_sms_sender_can_add_multiple_senders(client, notify_db_sessi
}
response = client.post('/service/{}/sms-sender'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
@@ -3150,7 +3150,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_updates_the_only_ex
}
response = client.post('/service/{}/sms-sender'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 201
updated_number = InboundNumber.query.get(inbound_number.id)
@@ -3176,7 +3176,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_inserts_new_sms_sen
}
response = client.post('/service/{}/sms-sender'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 201
updated_number = InboundNumber.query.get(inbound_number.id)
@@ -3198,7 +3198,7 @@ def test_add_service_sms_sender_switches_default(client, notify_db_session):
}
response = client.post('/service/{}/sms-sender'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
@@ -3216,7 +3216,7 @@ def test_add_service_sms_sender_return_404_when_service_does_not_exist(client):
}
response = client.post('/service/{}/sms-sender'.format(uuid.uuid4()),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 404
result = json.loads(response.get_data(as_text=True))
@@ -3233,7 +3233,7 @@ def test_update_service_sms_sender(client, notify_db_session):
}
response = client.post('/service/{}/sms-sender/{}'.format(service.id, service_sms_sender.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
@@ -3251,7 +3251,7 @@ def test_update_service_sms_sender_switches_default(client, notify_db_session):
}
response = client.post('/service/{}/sms-sender/{}'.format(service.id, service_sms_sender.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
@@ -3276,7 +3276,7 @@ def test_update_service_sms_sender_does_not_allow_sender_update_for_inbound_numb
}
response = client.post('/service/{}/sms-sender/{}'.format(service.id, service_sms_sender.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 400
@@ -3288,7 +3288,7 @@ def test_update_service_sms_sender_return_404_when_service_does_not_exist(client
}
response = client.post('/service/{}/sms-sender/{}'.format(uuid.uuid4(), uuid.uuid4()),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 404
result = json.loads(response.get_data(as_text=True))
@@ -3330,7 +3330,7 @@ def test_get_service_sms_sender_by_id(client, notify_db_session):
sms_sender='1235',
is_default=False)
response = client.get('/service/{}/sms-sender/{}'.format(service_sms_sender.service_id, service_sms_sender.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == service_sms_sender.serialize()
@@ -3341,7 +3341,7 @@ def test_get_service_sms_sender_by_id_returns_404_when_service_does_not_exist(cl
sms_sender='1235',
is_default=False)
response = client.get('/service/{}/sms-sender/{}'.format(uuid.uuid4(), service_sms_sender.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 404
@@ -3349,7 +3349,7 @@ def test_get_service_sms_sender_by_id_returns_404_when_service_does_not_exist(cl
def test_get_service_sms_sender_by_id_returns_404_when_sms_sender_does_not_exist(client, notify_db_session):
service = create_service()
response = client.get('/service/{}/sms-sender/{}'.format(service.id, uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 404
@@ -3359,7 +3359,7 @@ def test_get_service_sms_senders_for_service(client, notify_db_session):
sms_sender='second',
is_default=False)
response = client.get('/service/{}/sms-sender'.format(service_sms_sender.service_id),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
@@ -3372,7 +3372,7 @@ def test_get_service_sms_senders_for_service(client, notify_db_session):
def test_get_service_sms_senders_for_service_returns_empty_list_when_service_does_not_exist(client):
response = client.get('/service/{}/sms-sender'.format(uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == []

24
tests/app/service/test_service_data_retention_rest.py
View File

@@ -2,7 +2,7 @@ import json
import uuid
from app.models import ServiceDataRetention
from tests import create_authorization_header
from tests import create_admin_authorization_header
from tests.app.db import create_service_data_retention
@@ -15,7 +15,7 @@ def test_get_service_data_retention(client, sample_service):
response = client.get(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
)
assert response.status_code == 200
@@ -29,7 +29,7 @@ def test_get_service_data_retention(client, sample_service):
def test_get_service_data_retention_returns_empty_list(client, sample_service):
response = client.get(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
)
assert response.status_code == 200
assert len(json.loads(response.get_data(as_text=True))) == 0
@@ -38,7 +38,7 @@ def test_get_service_data_retention_returns_empty_list(client, sample_service):
def test_get_data_retention_for_service_notification_type(client, sample_service):
data_retention = create_service_data_retention(service=sample_service)
response = client.get('/service/{}/data-retention/notification-type/{}'.format(sample_service.id, 'sms'),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == data_retention.serialize()
@@ -52,7 +52,7 @@ def test_get_service_data_retention_by_id(client, sample_service):
days_of_retention=30)
response = client.get(
'/service/{}/data-retention/{}'.format(str(sample_service.id), sms_data_retention.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == sms_data_retention.serialize()
@@ -61,7 +61,7 @@ def test_get_service_data_retention_by_id(client, sample_service):
def test_get_service_data_retention_by_id_returns_none_when_no_data_retention_exists(client, sample_service):
response = client.get(
'/service/{}/data-retention/{}'.format(str(sample_service.id), uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {}
@@ -74,7 +74,7 @@ def test_create_service_data_retention(client, sample_service):
}
response = client.post(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
@@ -93,7 +93,7 @@ def test_create_service_data_retention_returns_400_when_notification_type_is_inv
}
response = client.post(
'/service/{}/data-retention'.format(str(uuid.uuid4())),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
json_resp = json.loads(response.get_data(as_text=True))
@@ -112,7 +112,7 @@ def test_create_service_data_retention_returns_400_when_data_retention_for_notif
}
response = client.post(
'/service/{}/data-retention'.format(str(uuid.uuid4())),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
@@ -129,7 +129,7 @@ def test_modify_service_data_retention(client, sample_service):
}
response = client.post(
'/service/{}/data-retention/{}'.format(sample_service.id, data_retention.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 204
@@ -142,7 +142,7 @@ def test_modify_service_data_retention_returns_400_when_data_retention_does_not_
}
response = client.post(
'/service/{}/data-retention/{}'.format(sample_service.id, uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
@@ -155,7 +155,7 @@ def test_modify_service_data_retention_returns_400_when_data_is_invalid(client):
}
response = client.post(
'/service/{}/data-retention/{}'.format(uuid.uuid4(), uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 400

16
tests/app/service/test_service_guest_list.py
View File

@@ -5,13 +5,13 @@ from app.dao.service_guest_list_dao import (
dao_add_and_commit_guest_list_contacts,
)
from app.models import EMAIL_TYPE, MOBILE_TYPE, ServiceGuestList
from tests import create_authorization_header
from tests import create_admin_authorization_header
def test_get_guest_list_returns_data(client, sample_service_guest_list):
service_id = sample_service_guest_list.service_id
response = client.get(f'service/{service_id}/guest-list', headers=[create_authorization_header()])
response = client.get(f'service/{service_id}/guest-list', headers=[create_admin_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {
'email_addresses': [sample_service_guest_list.recipient],
@@ -26,7 +26,9 @@ def test_get_guest_list_separates_emails_and_phones(client, sample_service):
ServiceGuestList.from_string(sample_service.id, MOBILE_TYPE, '+1800-555-555'),
])
response = client.get('service/{}/guest-list'.format(sample_service.id), headers=[create_authorization_header()])
response = client.get(
'service/{}/guest-list'.format(sample_service.id), headers=[create_admin_authorization_header()]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['email_addresses'] == ['service@example.com']
@@ -36,7 +38,7 @@ def test_get_guest_list_separates_emails_and_phones(client, sample_service):
def test_get_guest_list_404s_with_unknown_service_id(client):
path = 'service/{}/guest-list'.format(uuid.uuid4())
response = client.get(path, headers=[create_authorization_header()])
response = client.get(path, headers=[create_admin_authorization_header()])
assert response.status_code == 404
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
@@ -46,7 +48,7 @@ def test_get_guest_list_404s_with_unknown_service_id(client):
def test_get_guest_list_returns_no_data(client, sample_service):
path = 'service/{}/guest-list'.format(sample_service.id)
response = client.get(path, headers=[create_authorization_header()])
response = client.get(path, headers=[create_admin_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {'email_addresses': [], 'phone_numbers': []}
@@ -61,7 +63,7 @@ def test_update_guest_list_replaces_old_guest_list(client, sample_service_guest_
response = client.put(
f'service/{sample_service_guest_list.service_id}/guest-list',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 204
@@ -81,7 +83,7 @@ def test_update_guest_list_doesnt_remove_old_guest_list_if_error(client, sample_
response = client.put(
'service/{}/guest-list'.format(sample_service_guest_list.service_id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
headers=[('Content-Type', 'application/json'), create_admin_authorization_header()]
)
assert response.status_code == 400

16
tests/app/service/test_suspend_resume_service.py
View File

@@ -5,12 +5,12 @@ import pytest
from freezegun import freeze_time
from app.models import Service
from tests import create_authorization_header
from tests import create_admin_authorization_header
@pytest.mark.parametrize("endpoint", ["suspend", "resume"])
def test_only_allows_post(client, endpoint):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.get("/service/{}/{}".format(uuid.uuid4(), endpoint),
headers=[auth_header])
assert response.status_code == 405
@@ -18,7 +18,7 @@ def test_only_allows_post(client, endpoint):
@pytest.mark.parametrize("endpoint", ["suspend", "resume"])
def test_returns_404_when_service_does_not_exist(client, endpoint):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post("/service/{}/{}".format(uuid.uuid4(), endpoint),
headers=[auth_header])
assert response.status_code == 404
@@ -28,7 +28,7 @@ def test_returns_404_when_service_does_not_exist(client, endpoint):
def test_has_not_effect_when_service_is_already_that_state(client, sample_service, action, active, mocker):
mocked = mocker.patch("app.service.rest.dao_{}_service".format(action))
sample_service.active = active
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post("/service/{}/{}".format(sample_service.id, action),
headers=[auth_header])
assert response.status_code == 204
@@ -38,7 +38,7 @@ def test_has_not_effect_when_service_is_already_that_state(client, sample_servic
@freeze_time('2001-01-01T23:59:00')
def test_suspending_service_revokes_api_keys(client, sample_service, sample_api_key):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post("/service/{}/suspend".format(sample_service.id),
headers=[auth_header])
assert response.status_code == 204
@@ -47,11 +47,11 @@ def test_suspending_service_revokes_api_keys(client, sample_service, sample_api_
def test_resume_service_leaves_api_keys_revokes(client, sample_service, sample_api_key):
with freeze_time('2001-10-22T11:59:00'):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
client.post("/service/{}/suspend".format(sample_service.id),
headers=[auth_header])
with freeze_time('2001-10-22T13:59:00'):
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post("/service/{}/resume".format(sample_service.id),
headers=[auth_header])
assert response.status_code == 204
@@ -61,7 +61,7 @@ def test_resume_service_leaves_api_keys_revokes(client, sample_service, sample_a
@pytest.mark.parametrize("action, original_state", [("suspend", True), ("resume", False)])
def test_service_history_is_created(client, sample_service, action, original_state):
sample_service.active = original_state
auth_header = create_authorization_header()
auth_header = create_admin_authorization_header()
response = client.post("/service/{}/{}".format(sample_service.id, action),
headers=[auth_header])
ServiceHistory = Service.get_history_model()