import uuid import pytest from flask import current_app, json from freezegun import freeze_time from notifications_python_client.authentication import create_jwt_token from app.dao.api_key_dao import save_model_api_key from app.dao.notifications_dao import dao_update_notification from app.dao.templates_dao import dao_update_template from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST, ApiKey from tests import create_authorization_header from tests.app.db import create_api_key, create_notification @pytest.mark.parametrize('type', ('email', 'sms', 'letter')) def test_get_notification_by_id( client, sample_notification, sample_email_notification, sample_letter_notification, type ): if type == 'email': notification_to_get = sample_email_notification if type == 'sms': notification_to_get = sample_notification if type == 'letter': notification_to_get = sample_letter_notification auth_header = create_authorization_header(service_id=notification_to_get.service_id) response = client.get( '/notifications/{}'.format(notification_to_get.id), headers=[auth_header]) assert response.status_code == 200 notification = json.loads(response.get_data(as_text=True))['data']['notification'] assert notification['status'] == 'created' assert notification['template'] == { 'id': str(notification_to_get.template.id), 'name': notification_to_get.template.name, 'template_type': notification_to_get.template.template_type, 'version': 1 } assert notification['to'] == notification_to_get.to assert notification['service'] == str(notification_to_get.service_id) assert notification['body'] == notification_to_get.template.content assert notification.get('subject', None) == notification_to_get.subject @pytest.mark.parametrize("id", ["1234-badly-formatted-id-7890", "0"]) @pytest.mark.parametrize('type', ('email', 'sms')) def test_get_notification_by_invalid_id(client, sample_notification, sample_email_notification, id, type): if type == 'email': notification_to_get = sample_email_notification if type == 'sms': notification_to_get = sample_notification auth_header = create_authorization_header(service_id=notification_to_get.service_id) response = client.get( '/notifications/{}'.format(id), headers=[auth_header]) assert response.status_code == 405 def test_get_notifications_empty_result(client, sample_api_key): auth_header = create_authorization_header(service_id=sample_api_key.service_id) response = client.get( path='/notifications/{}'.format(uuid.uuid4()), headers=[auth_header]) notification = json.loads(response.get_data(as_text=True)) assert notification['result'] == "error" assert notification['message'] == "No result found" assert response.status_code == 404 @pytest.mark.parametrize('api_key_type,notification_key_type', [ (KEY_TYPE_NORMAL, KEY_TYPE_TEAM), (KEY_TYPE_NORMAL, KEY_TYPE_TEST), (KEY_TYPE_TEST, KEY_TYPE_NORMAL), (KEY_TYPE_TEST, KEY_TYPE_TEAM), (KEY_TYPE_TEAM, KEY_TYPE_NORMAL), (KEY_TYPE_TEAM, KEY_TYPE_TEST), ]) def test_get_notification_from_different_api_key_works( client, sample_notification, api_key_type, notification_key_type ): sample_notification.key_type = notification_key_type api_key = ApiKey(service=sample_notification.service, name='api_key', created_by=sample_notification.service.created_by, key_type=api_key_type) save_model_api_key(api_key) response = client.get( path='/notifications/{}'.format(sample_notification.id), headers=_create_auth_header_from_key(api_key)) assert response.status_code == 200 @pytest.mark.parametrize('key_type', [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST]) def test_get_notification_from_different_api_key_of_same_type_succeeds(client, sample_notification, key_type): creation_api_key = ApiKey(service=sample_notification.service, name='creation_api_key', created_by=sample_notification.service.created_by, key_type=key_type) save_model_api_key(creation_api_key) querying_api_key = ApiKey(service=sample_notification.service, name='querying_api_key', created_by=sample_notification.service.created_by, key_type=key_type) save_model_api_key(querying_api_key) sample_notification.api_key = creation_api_key sample_notification.key_type = key_type dao_update_notification(sample_notification) response = client.get( path='/notifications/{}'.format(sample_notification.id), headers=_create_auth_header_from_key(querying_api_key)) assert response.status_code == 200 notification = json.loads(response.get_data(as_text=True))['data']['notification'] assert sample_notification.api_key_id != querying_api_key.id assert notification['id'] == str(sample_notification.id) def test_get_all_notifications(client, sample_notification): auth_header = create_authorization_header(service_id=sample_notification.service_id) response = client.get( '/notifications', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert response.status_code == 200 assert notifications['notifications'][0]['status'] == 'created' assert notifications['notifications'][0]['template'] == { 'id': str(sample_notification.template.id), 'name': sample_notification.template.name, 'template_type': sample_notification.template.template_type, 'version': 1 } assert notifications['notifications'][0]['to'] == '+447700900855' assert notifications['notifications'][0]['service'] == str(sample_notification.service_id) assert notifications['notifications'][0]['body'] == 'Dear Sir/Madam, Hello. Yours Truly, The Government.' def test_normal_api_key_returns_notifications_created_from_jobs_and_from_api( client, sample_template, sample_api_key, sample_notification ): api_notification = create_notification(template=sample_template, api_key=sample_api_key) response = client.get( path='/notifications', headers=_create_auth_header_from_key(sample_api_key)) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True))['notifications'] assert len(notifications) == 2 assert set(x['id'] for x in notifications) == {str(sample_notification.id), str(api_notification.id)} @pytest.mark.parametrize('key_type', [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST]) def test_get_all_notifications_only_returns_notifications_of_matching_type( client, sample_template, sample_api_key, sample_test_api_key, sample_team_api_key, key_type ): normal_notification = create_notification( sample_template, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL ) team_notification = create_notification( sample_template, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM ) test_notification = create_notification( sample_template, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST ) notification_objs = { KEY_TYPE_NORMAL: normal_notification, KEY_TYPE_TEAM: team_notification, KEY_TYPE_TEST: test_notification } response = client.get( path='/notifications', headers=_create_auth_header_from_key(notification_objs[key_type].api_key)) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True))['notifications'] assert len(notifications) == 1 assert notifications[0]['id'] == str(notification_objs[key_type].id) @pytest.mark.parametrize('key_type', [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST]) def test_do_not_return_job_notifications_by_default( client, sample_template, sample_job, key_type ): team_api_key = create_api_key(sample_template.service, KEY_TYPE_TEAM) normal_api_key = create_api_key(sample_template.service, KEY_TYPE_NORMAL) test_api_key = create_api_key(sample_template.service, KEY_TYPE_TEST) create_notification(sample_template, job=sample_job) normal_notification = create_notification(sample_template, api_key=normal_api_key) team_notification = create_notification(sample_template, api_key=team_api_key) test_notification = create_notification(sample_template, api_key=test_api_key) notification_objs = { KEY_TYPE_NORMAL: normal_notification, KEY_TYPE_TEAM: team_notification, KEY_TYPE_TEST: test_notification } response = client.get( path='/notifications', headers=_create_auth_header_from_key(notification_objs[key_type].api_key)) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True))['notifications'] assert len(notifications) == 1 assert notifications[0]['id'] == str(notification_objs[key_type].id) @pytest.mark.parametrize('key_type', [ (KEY_TYPE_NORMAL, 2), (KEY_TYPE_TEAM, 1), (KEY_TYPE_TEST, 1) ]) def test_only_normal_api_keys_can_return_job_notifications( client, sample_notification_with_job, sample_template, sample_api_key, sample_team_api_key, sample_test_api_key, key_type ): normal_notification = create_notification( template=sample_template, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL ) team_notification = create_notification( template=sample_template, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM ) test_notification = create_notification( template=sample_template, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST ) notification_objs = { KEY_TYPE_NORMAL: normal_notification, KEY_TYPE_TEAM: team_notification, KEY_TYPE_TEST: test_notification } response = client.get( path='/notifications?include_jobs=true', headers=_create_auth_header_from_key(notification_objs[key_type[0]].api_key)) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True))['notifications'] assert len(notifications) == key_type[1] assert notifications[0]['id'] == str(notification_objs[key_type[0]].id) def test_get_all_notifications_newest_first(client, sample_email_template): notification_1 = create_notification(template=sample_email_template) notification_2 = create_notification(template=sample_email_template) notification_3 = create_notification(template=sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) 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_should_reject_invalid_page_param(client, sample_email_template): auth_header = create_authorization_header(service_id=sample_email_template.service_id) 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 'Not a valid integer.' in notifications['message']['page'] def test_valid_page_size_param(notify_api, sample_email_template): with notify_api.test_request_context(): create_notification(sample_email_template) create_notification(sample_email_template) with notify_api.test_client() as client: auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?page=1&page_size=1', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert response.status_code == 200 assert len(notifications['notifications']) == 1 assert notifications['total'] == 2 assert notifications['page_size'] == 1 def test_invalid_page_size_param(client, sample_email_template): create_notification(sample_email_template) create_notification(sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?page=1&page_size=invalid', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert response.status_code == 400 assert notifications['result'] == 'error' assert 'Not a valid integer.' in notifications['message']['page_size'] def test_should_return_pagination_links(client, sample_email_template): # Effectively mocking page size original_page_size = current_app.config['API_PAGE_SIZE'] try: current_app.config['API_PAGE_SIZE'] = 1 create_notification(sample_email_template) notification_2 = create_notification(sample_email_template) create_notification(sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) 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 finally: current_app.config['API_PAGE_SIZE'] = original_page_size def test_get_all_notifications_returns_empty_list(client, sample_api_key): auth_header = create_authorization_header(service_id=sample_api_key.service.id) 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_filter_by_template_type(client, sample_template, sample_email_template): create_notification(sample_template) create_notification(sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?template_type=sms', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert len(notifications['notifications']) == 1 assert notifications['notifications'][0]['template']['template_type'] == 'sms' assert response.status_code == 200 def test_filter_by_multiple_template_types(client, sample_template, sample_email_template): create_notification(sample_template) create_notification(sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?template_type=sms&template_type=email', headers=[auth_header]) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True)) assert len(notifications['notifications']) == 2 assert {'sms', 'email'} == set(x['template']['template_type'] for x in notifications['notifications']) def test_filter_by_status(client, sample_email_template): create_notification(sample_email_template, status="delivered") create_notification(sample_email_template) auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?status=delivered', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert len(notifications['notifications']) == 1 assert notifications['notifications'][0]['status'] == 'delivered' assert response.status_code == 200 def test_filter_by_multiple_statuses(client, sample_email_template): create_notification(sample_email_template, status="delivered") create_notification(sample_email_template, status='sending') auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?status=delivered&status=sending', headers=[auth_header] ) assert response.status_code == 200 notifications = json.loads(response.get_data(as_text=True)) assert len(notifications['notifications']) == 2 assert {'delivered', 'sending'} == set(x['status'] for x in notifications['notifications']) def test_filter_by_status_and_template_type(client, sample_template, sample_email_template): create_notification(sample_template) create_notification(sample_email_template) create_notification(sample_email_template, status="delivered") auth_header = create_authorization_header(service_id=sample_email_template.service_id) response = client.get( '/notifications?template_type=email&status=delivered', headers=[auth_header]) notifications = json.loads(response.get_data(as_text=True)) assert response.status_code == 200 assert len(notifications['notifications']) == 1 assert notifications['notifications'][0]['template']['template_type'] == 'email' assert notifications['notifications'][0]['status'] == 'delivered' def test_get_notification_by_id_returns_merged_template_content(client, sample_template_with_placeholders): sample_notification = create_notification(sample_template_with_placeholders, personalisation={"name": "world"}) auth_header = create_authorization_header(service_id=sample_notification.service_id) response = client.get( '/notifications/{}'.format(sample_notification.id), headers=[auth_header]) notification = json.loads(response.get_data(as_text=True))['data']['notification'] assert response.status_code == 200 assert notification['body'] == 'Hello world\nYour thing is due soon' assert 'subject' not in notification assert notification['content_char_count'] == 34 def test_get_notification_by_id_returns_merged_template_content_for_email( client, sample_email_template_with_placeholders ): sample_notification = create_notification( sample_email_template_with_placeholders, personalisation={"name": "world"} ) auth_header = create_authorization_header(service_id=sample_notification.service_id) response = client.get( '/notifications/{}'.format(sample_notification.id), headers=[auth_header]) notification = json.loads(response.get_data(as_text=True))['data']['notification'] assert response.status_code == 200 assert notification['body'] == 'Hello world\nThis is an email from GOV.UK' assert notification['subject'] == 'world' assert notification['content_char_count'] is None def test_get_notifications_for_service_returns_merged_template_content(client, sample_template_with_placeholders): with freeze_time('2001-01-01T12:00:00'): create_notification(sample_template_with_placeholders, personalisation={"name": "merged with first"}) with freeze_time('2001-01-01T12:00:01'): create_notification(sample_template_with_placeholders, personalisation={"name": "merged with second"}) auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id) response = client.get( path='/notifications', headers=[auth_header]) assert response.status_code == 200 assert {noti['body'] for noti in json.loads(response.get_data(as_text=True))['notifications']} == { 'Hello merged with first\nYour thing is due soon', 'Hello merged with second\nYour thing is due soon' } def test_get_notification_selects_correct_template_for_personalisation(client, notify_db, sample_template): create_notification(sample_template) original_content = sample_template.content sample_template.content = '((name))' dao_update_template(sample_template) notify_db.session.commit() create_notification(sample_template, personalisation={"name": "foo"}) auth_header = create_authorization_header(service_id=sample_template.service_id) response = client.get(path='/notifications', headers=[auth_header]) assert response.status_code == 200 resp = json.loads(response.get_data(as_text=True)) notis = sorted(resp['notifications'], key=lambda x: x['template_version']) assert len(notis) == 2 assert notis[0]['template_version'] == 1 assert notis[0]['body'] == original_content assert notis[1]['template_version'] == 2 assert notis[1]['body'] == 'foo' assert notis[0]['template_version'] == notis[0]['template']['version'] assert notis[1]['template_version'] == notis[1]['template']['version'] def _create_auth_header_from_key(api_key): token = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id)) return [('Authorization', 'Bearer {}'.format(token))]