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

More cleanup happening.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-16 15:49:46 -05:00
parent cf1db4dbf4
commit 287dcd6a46
6 changed files with 261 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
tests/app/notifications/test_process_notification.py
View File

@@ -11,7 +11,7 @@ from notifications_utils.recipients import (
)
from sqlalchemy.exc import SQLAlchemyError
from app.enums import ServicePermissionType, TemplateType
from app.enums import KeyType, NotificationType, ServicePermissionType, TemplateType
from app.models import Notification, NotificationHistory
from app.notifications.process_notifications import (
create_content_for_notification,
@@ -79,7 +79,7 @@ def test_persist_notification_creates_and_save_to_db(
recipient="+447111111111",
service=sample_template.service,
personalisation={},
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
@@ -123,7 +123,7 @@ def test_persist_notification_throws_exception_when_missing_template(sample_api_
recipient="+447111111111",
service=sample_api_key.service,
personalisation=None,
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
)
@@ -143,7 +143,7 @@ def test_persist_notification_with_optionals(sample_job, sample_api_key):
recipient="+12028675309",
service=sample_job.service,
personalisation=None,
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
created_at=created_at,
@@ -181,7 +181,7 @@ def test_persist_notification_cache_is_not_incremented_on_failure_to_create_noti
recipient="+447111111111",
service=sample_api_key.service,
personalisation=None,
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
)
@@ -191,20 +191,38 @@ def test_persist_notification_cache_is_not_incremented_on_failure_to_create_noti
@pytest.mark.parametrize(
("requested_queue, notification_type, key_type, expected_queue, expected_task"),
[
(None, "sms", "normal", "send-sms-tasks", "provider_tasks.deliver_sms"),
(None, "email", "normal", "send-email-tasks", "provider_tasks.deliver_email"),
(None, "sms", "team", "send-sms-tasks", "provider_tasks.deliver_sms"),
(
None,
NotificationType.SMS,
KeyType.NORMAL,
"send-sms-tasks",
"provider_tasks.deliver_sms",
),
(
None,
NotificationType.EMAIL,
KeyType.NORMAL,
"send-email-tasks",
"provider_tasks.deliver_email",
),
(
None,
NotificationType.SMS,
KeyType.TEAM,
"send-sms-tasks",
"provider_tasks.deliver_sms",
),
(
"notify-internal-tasks",
"sms",
"normal",
NotificationType.SMS,
KeyType.NORMAL,
"notify-internal-tasks",
"provider_tasks.deliver_sms",
),
(
"notify-internal-tasks",
"email",
"normal",
NotificationType.EMAIL,
KeyType.NORMAL,
"notify-internal-tasks",
"provider_tasks.deliver_email",
),
@@ -245,7 +263,8 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(
with pytest.raises(Boto3Error):
send_notification_to_queue(sample_notification, False)
mocked.assert_called_once_with(
[(str(sample_notification.id))], queue="send-sms-tasks"
[(str(sample_notification.id))],
queue="send-sms-tasks",
)
assert Notification.query.count() == 0
@@ -255,13 +274,13 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(
@pytest.mark.parametrize(
"to_address, notification_type, expected",
[
("+14254147755", "sms", True),
("+14254147167", "sms", True),
("simulate-delivered@notifications.service.gov.uk", "email", True),
("simulate-delivered-2@notifications.service.gov.uk", "email", True),
("simulate-delivered-3@notifications.service.gov.uk", "email", True),
("2028675309", "sms", False),
("valid_email@test.com", "email", False),
("+14254147755", NotificationType.SMS, True),
("+14254147167", NotificationType.SMS, True),
("simulate-delivered@notifications.service.gov.uk", NotificationType.EMAIL, True),
("simulate-delivered-2@notifications.service.gov.uk", NotificationType.EMAIL, True),
("simulate-delivered-3@notifications.service.gov.uk", NotificationType.EMAIL, True),
("2028675309", NotificationType.SMS, False),
("valid_email@test.com", NotificationType.EMAIL, False),
],
)
def test_simulated_recipient(notify_api, to_address, notification_type, expected):
@@ -277,7 +296,7 @@ def test_simulated_recipient(notify_api, to_address, notification_type, expected
"""
formatted_address = None
if notification_type == "email":
if notification_type == NotificationType.EMAIL:
formatted_address = validate_and_format_email_address(to_address)
else:
formatted_address = validate_and_format_phone_number(to_address)
@@ -311,7 +330,7 @@ def test_persist_notification_with_international_info_stores_correct_info(
recipient=recipient,
service=sample_job.service,
personalisation=None,
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
@@ -334,7 +353,7 @@ def test_persist_notification_with_international_info_does_not_store_for_email(
recipient="foo@bar.com",
service=sample_job.service,
personalisation=None,
notification_type="email",
notification_type=NotificationType.EMAIL,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
@@ -368,7 +387,7 @@ def test_persist_sms_notification_stores_normalised_number(
recipient=recipient,
service=sample_job.service,
personalisation=None,
notification_type="sms",
notification_type=NotificationType.SMS,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
@@ -392,7 +411,7 @@ def test_persist_email_notification_stores_normalised_email(
recipient=recipient,
service=sample_job.service,
personalisation=None,
notification_type="email",
notification_type=NotificationType.EMAIL,
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
@@ -415,7 +434,7 @@ def test_persist_notification_with_billable_units_stores_correct_info(mocker):
personalisation=None,
notification_type=template.template_type,
api_key_id=None,
key_type="normal",
key_type=KeyType.NORMAL,
billable_units=3,
)
persisted_notification = Notification.query.all()[0]

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

@@ -8,19 +8,19 @@ 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.enums import KeyType
from app.enums import KeyType, NotificationStatus, NotificationType, TemplateType
from app.models import ApiKey
from tests import create_service_authorization_header
from tests.app.db import create_api_key, create_notification
@pytest.mark.parametrize("type", ("email", "sms"))
@pytest.mark.parametrize("type", (NotificationType.EMAIL, NotificationType.SMS))
def test_get_notification_by_id(
client, sample_notification, sample_email_notification, type
):
if type == "email":
if type == NotificationType.EMAIL:
notification_to_get = sample_email_notification
if type == "sms":
elif type == NotificationType.SMS:
notification_to_get = sample_notification
auth_header = create_service_authorization_header(
@@ -46,13 +46,13 @@ def test_get_notification_by_id(
@pytest.mark.parametrize("id", ["1234-badly-formatted-id-7890", "0"])
@pytest.mark.parametrize("type", ("email", "sms"))
@pytest.mark.parametrize("type", (NotificationType.EMAIL, NotificationType.SMS))
def test_get_notification_by_invalid_id(
client, sample_notification, sample_email_notification, id, type
):
if type == "email":
if type == NotificationType.EMAIL:
notification_to_get = sample_email_notification
if type == "sms":
elif type == NotificationType.SMS:
notification_to_get = sample_notification
auth_header = create_service_authorization_header(
service_id=notification_to_get.service_id
@@ -420,7 +420,10 @@ def test_filter_by_template_type(client, sample_template, sample_email_template)
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications["notifications"]) == 1
assert notifications["notifications"][0]["template"]["template_type"] == "sms"
assert (
notifications["notifications"][0]["template"]["template_type"]
== TemplateType.SMS
)
assert response.status_code == 200
@@ -441,13 +444,13 @@ def test_filter_by_multiple_template_types(
assert response.status_code == 200
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications["notifications"]) == 2
assert {"sms", "email"} == set(
assert {TemplateType.SMS, TemplateType.EMAIL} == {
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, status=NotificationStatus.DELIVERED)
create_notification(sample_email_template)
auth_header = create_service_authorization_header(
@@ -458,13 +461,13 @@ def test_filter_by_status(client, sample_email_template):
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications["notifications"]) == 1
assert notifications["notifications"][0]["status"] == "delivered"
assert notifications["notifications"][0]["status"] == NotificationStatus.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")
create_notification(sample_email_template, status=NotificationStatus.DELIVERED)
create_notification(sample_email_template, status=NotificationStatus.SENDING)
auth_header = create_service_authorization_header(
service_id=sample_email_template.service_id
@@ -477,9 +480,9 @@ def test_filter_by_multiple_statuses(client, sample_email_template):
assert response.status_code == 200
notifications = json.loads(response.get_data(as_text=True))
assert len(notifications["notifications"]) == 2
assert {"delivered", "sending"} == set(
assert {NotificationStatus.DELIVERED, NotificationStatus.SENDING} == {
x["status"] for x in notifications["notifications"]
)
}
def test_filter_by_status_and_template_type(
@@ -487,7 +490,7 @@ def test_filter_by_status_and_template_type(
):
create_notification(sample_template)
create_notification(sample_email_template)
create_notification(sample_email_template, status="delivered")
create_notification(sample_email_template, status=NotificationStatus.DELIVERED)
auth_header = create_service_authorization_header(
service_id=sample_email_template.service_id
@@ -500,8 +503,11 @@ def test_filter_by_status_and_template_type(
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"
assert (
notifications["notifications"][0]["template"]["template_type"]
== TemplateType.EMAIL
)
assert notifications["notifications"][0]["status"] == NotificationStatus.DELIVERED
def test_get_notification_by_id_returns_merged_template_content(

103
tests/app/notifications/test_validators.py
View File

@@ -54,7 +54,7 @@ def enable_redis(notify_api):
yield
@pytest.mark.parametrize("key_type", ["team", "normal"])
@pytest.mark.parametrize("key_type", [KeyType.TEAM, KeyType.NORMAL])
def test_check_service_over_total_message_limit_fails(
key_type, mocker, notify_db_session
):
@@ -71,7 +71,7 @@ def test_check_service_over_total_message_limit_fails(
assert e.value.fields == []
@pytest.mark.parametrize("key_type", ["team", "normal"])
@pytest.mark.parametrize("key_type", [KeyType.TEAM, KeyType.NORMAL])
def test_check_application_over_retention_limit_fails(
key_type, mocker, notify_db_session
):
@@ -142,7 +142,7 @@ def test_check_template_is_active_fails(sample_template):
assert e.value.fields == [{"template": "Template has been deleted"}]
@pytest.mark.parametrize("key_type", ["test", "normal"])
@pytest.mark.parametrize("key_type", [KeyType.TEST, KeyType.NORMAL])
def test_service_can_send_to_recipient_passes(key_type, notify_db_session):
trial_mode_service = create_service(service_name="trial mode", restricted=True)
serialised_service = SerialisedService.from_id(trial_mode_service.id)
@@ -175,7 +175,11 @@ def test_service_can_send_to_recipient_passes_with_non_normalized_number(
serialised_service = SerialisedService.from_id(sample_service.id)
assert (
service_can_send_to_recipient(recipient_number, "team", serialised_service)
service_can_send_to_recipient(
recipient_number,
KeyType.TEAM,
serialised_service,
)
is None
)
@@ -194,12 +198,12 @@ def test_service_can_send_to_recipient_passes_with_non_normalized_email(
serialised_service = SerialisedService.from_id(sample_service.id)
assert (
service_can_send_to_recipient(recipient_email, "team", serialised_service)
service_can_send_to_recipient(recipient_email, KeyType.TEAM, serialised_service)
is None
)
@pytest.mark.parametrize("key_type", ["test", "normal"])
@pytest.mark.parametrize("key_type", [KeyType.TEST, KeyType.NORMAL])
def test_service_can_send_to_recipient_passes_for_live_service_non_team_member(
key_type, sample_service
):
@@ -222,12 +226,19 @@ def test_service_can_send_to_recipient_passes_for_guest_list_recipient_passes(
create_service_guest_list(sample_service, email_address="some_other_email@test.com")
assert (
service_can_send_to_recipient(
"some_other_email@test.com", "team", sample_service
"some_other_email@test.com", KeyType.TEAM, sample_service
)
is None
)
create_service_guest_list(sample_service, mobile_number="2028675309")
assert service_can_send_to_recipient("2028675309", "team", sample_service) is None
assert (
service_can_send_to_recipient(
"2028675309",
KeyType.TEAM,
sample_service,
)
is None
)
@pytest.mark.parametrize(
@@ -246,7 +257,7 @@ def test_service_can_send_to_recipient_fails_when_ignoring_guest_list(
with pytest.raises(BadRequestError) as exec_info:
service_can_send_to_recipient(
next(iter(recipient.values())),
"team",
KeyType.TEAM,
sample_service,
allow_guest_list_recipients=False,
)
@@ -262,9 +273,9 @@ def test_service_can_send_to_recipient_fails_when_ignoring_guest_list(
@pytest.mark.parametrize(
"key_type, error_message",
[
("team", "Can’t send to this recipient using a team-only API key"),
(KeyType.TEAM, "Can’t send to this recipient using a team-only API key"),
(
"normal",
KeyType.NORMAL,
"Can’t send to this recipient when service is in trial mode – see https://www.notifications.service.gov.uk/trial-mode", # noqa
),
],
@@ -287,7 +298,7 @@ def test_service_can_send_to_recipient_fails_when_mobile_number_is_not_on_team(
sample_service,
):
with pytest.raises(BadRequestError) as e:
service_can_send_to_recipient("0758964221", "team", sample_service)
service_can_send_to_recipient("0758964221", KeyType.TEAM, sample_service)
assert e.value.status_code == 400
assert e.value.message == "Can’t send to this recipient using a team-only API key"
assert e.value.fields == []
@@ -295,7 +306,7 @@ def test_service_can_send_to_recipient_fails_when_mobile_number_is_not_on_team(
@pytest.mark.parametrize("char_count", [612, 0, 494, 200, 918])
@pytest.mark.parametrize("show_prefix", [True, False])
@pytest.mark.parametrize("template_type", ["sms", "email"])
@pytest.mark.parametrize("template_type", [TemplateType.SMS, TemplateType.EMAIL])
def test_check_is_message_too_long_passes(
notify_db_session, show_prefix, char_count, template_type
):
@@ -316,7 +327,7 @@ def test_check_is_message_too_long_fails(notify_db_session, show_prefix, char_co
with pytest.raises(BadRequestError) as e:
service = create_service(prefix_sms=show_prefix)
t = create_template(
service=service, content="a" * char_count, template_type="sms"
service=service, content="a" * char_count, template_type=TemplateType.SMS
)
template = templates_dao.dao_get_template_by_id_and_service_id(
template_id=t.id, service_id=service.id
@@ -340,7 +351,7 @@ def test_check_is_message_too_long_passes_for_long_email(sample_service):
t = create_template(
service=sample_service,
content="a" * email_character_count,
template_type="email",
template_type=TemplateType.EMAIL,
)
template = templates_dao.dao_get_template_by_id_and_service_id(
template_id=t.id, service_id=t.service_id
@@ -392,15 +403,15 @@ def test_check_notification_content_is_not_empty_fails(
def test_validate_template(sample_service):
template = create_template(sample_service, template_type="email")
validate_template(template.id, {}, sample_service, "email")
template = create_template(sample_service, template_type=TemplateType.EMAIL)
validate_template(template.id, {}, sample_service, NotificationType.EMAIL)
@pytest.mark.parametrize("check_char_count", [True, False])
def test_validate_template_calls_all_validators(
mocker, fake_uuid, sample_service, check_char_count
):
template = create_template(sample_service, template_type="email")
template = create_template(sample_service, template_type=TemplateType.EMAIL)
mock_check_type = mocker.patch(
"app.notifications.validators.check_template_is_for_notification_type"
)
@@ -418,10 +429,14 @@ def test_validate_template_calls_all_validators(
"app.notifications.validators.check_is_message_too_long"
)
template, template_with_content = validate_template(
template.id, {}, sample_service, "email", check_char_count=check_char_count
template.id,
{},
sample_service,
NotificationType.EMAIL,
check_char_count=check_char_count,
)
mock_check_type.assert_called_once_with("email", "email")
mock_check_type.assert_called_once_with(NotificationType.EMAIL, TemplateType.EMAIL)
mock_check_if_active.assert_called_once_with(template)
mock_create_conent.assert_called_once_with(template, {})
mock_check_not_empty.assert_called_once_with("content")
@@ -434,7 +449,7 @@ def test_validate_template_calls_all_validators(
def test_validate_template_calls_all_validators_exception_message_too_long(
mocker, fake_uuid, sample_service
):
template = create_template(sample_service, template_type="email")
template = create_template(sample_service, template_type=TemplateType.EMAIL)
mock_check_type = mocker.patch(
"app.notifications.validators.check_template_is_for_notification_type"
)
@@ -452,7 +467,11 @@ def test_validate_template_calls_all_validators_exception_message_too_long(
"app.notifications.validators.check_is_message_too_long"
)
template, template_with_content = validate_template(
template.id, {}, sample_service, "email", check_char_count=False
template.id,
{},
sample_service,
NotificationType.EMAIL,
check_char_count=False,
)
mock_check_type.assert_called_once_with("email", "email")
@@ -462,20 +481,15 @@ def test_validate_template_calls_all_validators_exception_message_too_long(
assert not mock_check_message_is_too_long.called
@pytest.mark.parametrize("key_type", ["team", "live", "test"])
@pytest.mark.parametrize("key_type", [KeyType.TEAM, KeyType.NORMAL, KeyType.TEST])
def test_check_service_over_api_rate_limit_when_exceed_rate_limit_request_fails_raises_error(
key_type, sample_service, mocker
):
with freeze_time("2016-01-01 12:00:00.000000"):
if key_type == "live":
api_key_type = "normal"
else:
api_key_type = key_type
mocker.patch("app.redis_store.exceeded_rate_limit", return_value=True)
sample_service.restricted = True
api_key = create_api_key(sample_service, key_type=api_key_type)
api_key = create_api_key(sample_service, key_type=key_type)
serialised_service = SerialisedService.from_id(sample_service.id)
serialised_api_key = SerialisedAPIKeyCollection.from_service_id(
serialised_service.id
@@ -490,17 +504,16 @@ def test_check_service_over_api_rate_limit_when_exceed_rate_limit_request_fails_
60,
)
assert e.value.status_code == 429
assert (
e.value.message
== "Exceeded rate limit for key type {} of {} requests per {} seconds".format(
key_type.upper(), sample_service.rate_limit, 60
)
assert e.value.message == (
f"Exceeded rate limit for key type {key_type.upper()} of "
f"{sample_service.rate_limit} requests per {60} seconds"
)
assert e.value.fields == []
def test_check_service_over_api_rate_limit_when_rate_limit_has_not_exceeded_limit_succeeds(
sample_service, mocker
sample_service,
mocker,
):
with freeze_time("2016-01-01 12:00:00.000000"):
mocker.patch("app.redis_store.exceeded_rate_limit", return_value=False)
@@ -551,7 +564,7 @@ def test_check_rate_limiting_validates_api_rate_limit_and_daily_limit(
mock_rate_limit.assert_called_once_with(service, api_key)
@pytest.mark.parametrize("key_type", ["test", "normal"])
@pytest.mark.parametrize("key_type", [KeyType.TEST, KeyType.NORMAL])
def test_validate_and_format_recipient_fails_when_international_number_and_service_does_not_allow_int_sms(
key_type,
notify_db_session,
@@ -590,7 +603,10 @@ def test_validate_and_format_recipient_fails_when_no_recipient():
assert e.value.message == "Recipient can't be empty"
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize(
"notification_type",
[NotificationType.SMS, NotificationType.EMAIL],
)
def test_check_service_email_reply_to_id_where_reply_to_id_is_none(notification_type):
assert check_service_email_reply_to_id(None, None, notification_type) is None
@@ -638,7 +654,10 @@ def test_check_service_email_reply_to_id_where_reply_to_id_is_not_found(
)
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize(
"notification_type",
[NotificationType.SMS, NotificationType.EMAIL],
)
def test_check_service_sms_sender_id_where_sms_sender_id_is_none(notification_type):
assert check_service_sms_sender_id(None, None, notification_type) is None
@@ -684,7 +703,10 @@ def test_check_service_sms_sender_id_where_sms_sender_is_not_found(
)
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize(
"notification_type",
[NotificationType.SMS, NotificationType.EMAIL],
)
def test_check_reply_to_with_empty_reply_to(sample_service, notification_type):
assert check_reply_to(sample_service.id, None, notification_type) is None
@@ -778,6 +800,7 @@ def test_check_service_over_total_message_limit(mocker, sample_service):
get_redis_mock = mocker.patch("app.notifications.validators.redis_store.get")
get_redis_mock.return_value = None
service_stats = check_service_over_total_message_limit(
KeyType.NORMAL, sample_service
KeyType.NORMAL,
sample_service,
)
assert service_stats == 0

9
tests/app/performance_dashboard/test_rest.py
View File

@@ -1,5 +1,6 @@
from datetime import date
from app.enums import TemplateType
from tests.app.db import (
create_ft_notification_status,
create_process_time,
@@ -9,10 +10,14 @@ from tests.app.db import (
def test_performance_dashboard(sample_service, admin_request):
template_sms = create_template(
service=sample_service, template_type="sms", template_name="a"
service=sample_service,
template_type=TemplateType.SMS,
template_name="a",
)
template_email = create_template(
service=sample_service, template_type="email", template_name="b"
service=sample_service,
template_type=TemplateType.EMAIL,
template_name="b",
)
create_ft_notification_status(
local_date=date(2021, 2, 28),

22
tests/app/platform_stats/test_rest.py
View File

@@ -3,7 +3,7 @@ from datetime import date, datetime
import pytest
from freezegun import freeze_time
from app.enums import TemplateType
from app.enums import KeyType, NotificationStatus, NotificationType, TemplateType
from app.errors import InvalidRequest
from app.platform_stats.rest import validate_date_range_is_within_a_financial_year
from tests.app.db import (
@@ -64,17 +64,27 @@ def test_get_platform_stats_with_real_query(admin_request, notify_db_session):
service=service_1,
template_type=TemplateType.EMAIL,
)
create_ft_notification_status(date(2018, 10, 29), "sms", service_1, count=10)
create_ft_notification_status(date(2018, 10, 29), "email", service_1, count=3)
create_ft_notification_status(
date(2018, 10, 29), NotificationType.SMS, service_1, count=10
)
create_ft_notification_status(
date(2018, 10, 29), NotificationType.EMAIL, service_1, count=3
)
create_notification(
sms_template, created_at=datetime(2018, 10, 31, 11, 0, 0), key_type="test"
sms_template,
created_at=datetime(2018, 10, 31, 11, 0, 0),
key_type=KeyType.TEST,
)
create_notification(
sms_template, created_at=datetime(2018, 10, 31, 12, 0, 0), status="delivered"
sms_template,
created_at=datetime(2018, 10, 31, 12, 0, 0),
status=NotificationStatus.DELIVERED,
)
create_notification(
email_template, created_at=datetime(2018, 10, 31, 13, 0, 0), status="delivered"
email_template,
created_at=datetime(2018, 10, 31, 13, 0, 0),
status=NotificationStatus.DELIVERED,
)
response = admin_request.get(

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

@@ -16,6 +16,7 @@ from app.dao.templates_dao import dao_redact_template
from app.dao.users_dao import save_model_user
from app.enums import (
KeyType,
NotificationStatus,
NotificationType,
OrganizationType,
PermissionType,
@@ -136,9 +137,10 @@ def test_find_services_by_name_finds_services(notify_db_session, admin_request,
"app.service.rest.get_services_by_partial_name",
return_value=[service_1, service_2],
)
response = admin_request.get("service.find_services_by_name", service_name="ABC")[
"data"
]
response = admin_request.get(
"service.find_services_by_name",
service_name="ABC",
)["data"]
mock_get_services_by_partial_name.assert_called_once_with("ABC")
assert len(response) == 2
@@ -172,11 +174,13 @@ def test_get_live_services_data(sample_user, admin_request):
service = create_service(go_live_user=sample_user, go_live_at=datetime(2018, 1, 1))
service_2 = create_service(
service_name="second", go_live_at=datetime(2019, 1, 1), go_live_user=sample_user
service_name="second",
go_live_at=datetime(2019, 1, 1),
go_live_user=sample_user,
)
sms_template = create_template(service=service)
email_template = create_template(service=service, template_type="email")
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
dao_add_service_to_organization(service=service, organization_id=org.id)
create_ft_billing(local_date="2019-04-20", template=sms_template)
create_ft_billing(local_date="2019-04-20", template=email_template)
@@ -269,7 +273,9 @@ def test_get_service_by_id_returns_organization_type(
admin_request, sample_service, detailed
):
json_resp = admin_request.get(
"service.get_service_by_id", service_id=sample_service.id, detailed=detailed
"service.get_service_by_id",
service_id=sample_service.id,
detailed=detailed,
)
assert json_resp["data"]["organization_type"] is None
@@ -297,7 +303,8 @@ def test_get_service_by_id_has_default_service_permissions(
admin_request, sample_service
):
json_resp = admin_request.get(
"service.get_service_by_id", service_id=sample_service.id
"service.get_service_by_id",
service_id=sample_service.id,
)
assert set(json_resp["data"]["permissions"]) == {
@@ -309,7 +316,9 @@ def test_get_service_by_id_has_default_service_permissions(
def test_get_service_by_id_should_404_if_no_service(admin_request, notify_db_session):
json_resp = admin_request.get(
"service.get_service_by_id", service_id=uuid.uuid4(), _expected_status=404
"service.get_service_by_id",
service_id=uuid.uuid4(),
_expected_status=404,
)
assert json_resp["result"] == "error"
@@ -381,7 +390,9 @@ def test_create_service(
}
json_resp = admin_request.post(
"service.create_service", _data=data, _expected_status=201
"service.create_service",
_data=data,
_expected_status=201,
)
assert json_resp["data"]["id"]
@@ -452,7 +463,9 @@ def test_create_service_with_domain_sets_organization(
}
json_resp = admin_request.post(
"service.create_service", _data=data, _expected_status=201
"service.create_service",
_data=data,
_expected_status=201,
)
if expected_org:
@@ -679,7 +692,9 @@ def test_create_service_should_throw_duplicate_key_constraint_for_existing_email
def test_update_service(client, notify_db_session, sample_service):
brand = EmailBranding(
colour="#000000", logo="justice-league.png", name="Justice League"
colour="#000000",
logo="justice-league.png",
name="Justice League",
)
notify_db_session.add(brand)
notify_db_session.commit()
@@ -731,7 +746,9 @@ def test_update_service_remove_email_branding(
admin_request, notify_db_session, sample_service
):
brand = EmailBranding(
colour="#000000", logo="justice-league.png", name="Justice League"
colour="#000000",
logo="justice-league.png",
name="Justice League",
)
sample_service.email_branding = brand
notify_db_session.commit()
@@ -748,7 +765,9 @@ def test_update_service_change_email_branding(
admin_request, notify_db_session, sample_service
):
brand1 = EmailBranding(
colour="#000000", logo="justice-league.png", name="Justice League"
colour="#000000",
logo="justice-league.png",
name="Justice League",
)
brand2 = EmailBranding(colour="#111111", logo="avengers.png", name="Avengers")
notify_db_session.add_all([brand1, brand2])
@@ -1251,13 +1270,13 @@ def test_add_existing_user_to_another_service_with_all_permissions(
data = {
"permissions": [
{"permission": "send_emails"},
{"permission": "send_texts"},
{"permission": "manage_users"},
{"permission": "manage_settings"},
{"permission": "manage_api_keys"},
{"permission": "manage_templates"},
{"permission": "view_activity"},
{"permission": PermissionType.SEND_EMAILS},
{"permission": PermissionType.SEND_TEXTS},
{"permission": PermissionType.MANAGE_USERS},
{"permission": PermissionType.MANAGE_SETTINGS},
{"permission": PermissionType.MANAGE_API_KEYS},
{"permission": PermissionType.MANAGE_TEMPLATES},
{"permission": PermissionType.VIEW_ACTIVITY},
],
"folder_permissions": [],
}
@@ -1320,8 +1339,8 @@ def test_add_existing_user_to_another_service_with_send_permissions(
data = {
"permissions": [
{"permission": "send_emails"},
{"permission": "send_texts"},
{"permission": PermissionType.SEND_EMAILS},
{"permission": PermissionType.SEND_TEXTS},
],
"folder_permissions": [],
}
@@ -1367,9 +1386,9 @@ def test_add_existing_user_to_another_service_with_manage_permissions(
data = {
"permissions": [
{"permission": "manage_users"},
{"permission": "manage_settings"},
{"permission": "manage_templates"},
{"permission": PermissionType.MANAGE_USERS},
{"permission": PermissionType.MANAGE_SETTINGS},
{"permission": PermissionType.MANAGE_TEMPLATES},
]
}
@@ -1395,9 +1414,9 @@ def test_add_existing_user_to_another_service_with_manage_permissions(
permissions = json_resp["data"]["permissions"][str(sample_service.id)]
expected_permissions = [
"manage_users",
"manage_settings",
"manage_templates",
PermissionType.MANAGE_USERS,
PermissionType.MANAGE_SETTINGS,
PermissionType.MANAGE_TEMPLATES,
]
assert sorted(expected_permissions) == sorted(permissions)
@@ -1420,7 +1439,7 @@ def test_add_existing_user_to_another_service_with_folder_permissions(
folder_2 = create_template_folder(sample_service)
data = {
"permissions": [{"permission": "manage_api_keys"}],
"permissions": [{"permission": PermissionType.MANAGE_API_KEYS}],
"folder_permissions": [str(folder_1.id), str(folder_2.id)],
}
@@ -1457,7 +1476,7 @@ def test_add_existing_user_to_another_service_with_manage_api_keys(
)
save_model_user(user_to_add, validated_email_access=True)
data = {"permissions": [{"permission": "manage_api_keys"}]}
data = {"permissions": [{"permission": PermissionType.MANAGE_API_KEYS}]}
auth_header = create_admin_authorization_header()
@@ -1748,7 +1767,7 @@ def test_get_all_notifications_for_service_filters_notifications_when_using_post
auth_header = create_admin_authorization_header()
data = {
"page": 1,
"template_type": ["sms"],
"template_type": [TemplateType.SMS],
"status": ["created", "sending"],
"to": "0855",
}
@@ -2059,7 +2078,9 @@ def test_set_sms_prefixing_for_service_cant_be_none(
def test_get_detailed_service(
sample_template, client, sample_service, today_only, stats
):
create_ft_notification_status(date(2000, 1, 1), "sms", sample_service, count=1)
create_ft_notification_status(
date(2000, 1, 1), NotificationType.SMS, sample_service, count=1
)
with freeze_time("2000-01-02T12:00:00"):
create_notification(template=sample_template, status="created")
resp = client.get(
@@ -2259,21 +2280,23 @@ def test_get_detailed_services_for_date_range(
create_ft_notification_status(
local_date=(datetime.utcnow() - timedelta(days=3)).date(),
service=sample_template.service,
notification_type="sms",
notification_type=NotificationType.SMS,
)
create_ft_notification_status(
local_date=(datetime.utcnow() - timedelta(days=2)).date(),
service=sample_template.service,
notification_type="sms",
notification_type=NotificationType.SMS,
)
create_ft_notification_status(
local_date=(datetime.utcnow() - timedelta(days=1)).date(),
service=sample_template.service,
notification_type="sms",
notification_type=NotificationType.SMS,
)
create_notification(
template=sample_template, created_at=datetime.utcnow(), status="delivered"
template=sample_template,
created_at=datetime.utcnow(),
status=NotificationStatus.DELIVERED,
)
start_date = (datetime.utcnow() - timedelta(days=start_date_delta)).date()
@@ -2337,9 +2360,8 @@ def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_ma
create_notification(sample_email_template, to_field="jack@gmail.com")
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
notification1.service_id, "+447700900800", "sms"
),
f"/service/{notification1.service_id}/notifications?"
f"to={+447700900800}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2368,9 +2390,8 @@ def test_search_for_notification_by_to_field_return_multiple_matches(
)
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
notification1.service_id, "+447700900855", "sms"
),
f"/service/{notification1.service_id}/notifications?"
f"to={+447700900855}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2397,9 +2418,8 @@ 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"
),
f"/service/{sample_template.service_id}/notifications?"
f"to={+447700900855}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
assert response.status_code == 200
@@ -2422,9 +2442,8 @@ def test_search_for_notification_by_to_field_returns_no_next_link_if_50_or_less(
)
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
sample_template.service_id, "+447700900855", "sms"
),
f"/service/{sample_template.service_id}/notifications?"
f"to={+447700900855}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
assert response.status_code == 200
@@ -2447,7 +2466,7 @@ def test_update_service_calls_send_notification_as_service_becomes_live(
auth_header = create_admin_authorization_header()
resp = client.post(
"service/{}".format(restricted_service.id),
f"service/{restricted_service.id}",
data=json.dumps(data),
headers=[auth_header],
content_type="application/json",
@@ -2494,7 +2513,7 @@ def test_update_service_does_not_call_send_notification_when_restricted_not_chan
auth_header = create_admin_authorization_header()
resp = client.post(
"service/{}".format(sample_service.id),
f"service/{sample_service.id}",
data=json.dumps(data),
headers=[auth_header],
content_type="application/json",
@@ -2522,9 +2541,8 @@ def test_search_for_notification_by_to_field_filters_by_status(client, sample_te
)
response = client.get(
"/service/{}/notifications?to={}&status={}&template_type={}".format(
notification1.service_id, "+447700900855", "delivered", "sms"
),
f"/service/{notification1.service_id}/notifications?to={+447700900855}"
f"&status={NotificationStatus.DELIVERED}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2555,9 +2573,9 @@ def test_search_for_notification_by_to_field_filters_by_statuses(
)
response = client.get(
"/service/{}/notifications?to={}&status={}&status={}&template_type={}".format(
notification1.service_id, "+447700900855", "delivered", "sending", "sms"
),
f"/service/{notification1.service_id}/notifications?to={+447700900855}"
f"&status={NotificationStatus.DELIVERED}&status={NotificationStatus.SENDING}"
f"&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2583,9 +2601,8 @@ def test_search_for_notification_by_to_field_returns_content(
)
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
sample_template_with_placeholders.service_id, "+447700900855", "sms"
),
f"/service/{sample_template_with_placeholders.service_id}/notifications?"
f"to={+447700900855}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2689,9 +2706,8 @@ def test_search_for_notification_by_to_field_returns_personlisation(
)
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
sample_template_with_placeholders.service_id, "+447700900855", "sms"
),
f"/service/{sample_template_with_placeholders.service_id}/notifications?"
f"to={+447700900855}&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2709,7 +2725,9 @@ def test_search_for_notification_by_to_field_returns_notifications_by_type(
client, sample_template, sample_email_template
):
sms_notification = create_notification(
sample_template, to_field="+447700900855", normalised_to="447700900855"
sample_template,
to_field="+447700900855",
normalised_to="447700900855",
)
create_notification(
sample_email_template,
@@ -2718,9 +2736,8 @@ def test_search_for_notification_by_to_field_returns_notifications_by_type(
)
response = client.get(
"/service/{}/notifications?to={}&template_type={}".format(
sms_notification.service_id, "0770", "sms"
),
f"/service/{sms_notification.service_id}/notifications?to={'0770'}"
f"&template_type={TemplateType.SMS}",
headers=[create_admin_authorization_header()],
)
notifications = json.loads(response.get_data(as_text=True))["notifications"]
@@ -2734,7 +2751,7 @@ 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),
f"/service/{sample_service.id}/email-reply-to",
headers=[create_admin_authorization_header()],
)
@@ -2747,7 +2764,7 @@ def test_get_email_reply_to_addresses_with_one_email_address(client, notify_db_s
create_reply_to_email(service, "test@mail.com")
response = client.get(
"/service/{}/email-reply-to".format(service.id),
f"/service/{service.id}/email-reply-to",
headers=[create_admin_authorization_header()],
)
json_response = json.loads(response.get_data(as_text=True))
@@ -2768,7 +2785,7 @@ def test_get_email_reply_to_addresses_with_multiple_email_addresses(
reply_to_b = create_reply_to_email(service, "test_b@mail.com", False)
response = client.get(
"/service/{}/email-reply-to".format(service.id),
f"/service/{service.id}/email-reply-to",
headers=[create_admin_authorization_header()],
)
json_response = json.loads(response.get_data(as_text=True))
@@ -3065,7 +3082,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_updates_the_only_ex
"inbound_number_id": str(inbound_number.id),
}
response = client.post(
"/service/{}/sms-sender".format(service.id),
f"/service/{service.id}/sms-sender",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3096,7 +3113,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_inserts_new_sms_sen
"inbound_number_id": str(inbound_number.id),
}
response = client.post(
"/service/{}/sms-sender".format(service.id),
f"/service/{service.id}/sms-sender",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3122,7 +3139,7 @@ def test_add_service_sms_sender_switches_default(client, notify_db_session):
"is_default": True,
}
response = client.post(
"/service/{}/sms-sender".format(service.id),
f"/service/{service.id}/sms-sender",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3141,7 +3158,7 @@ def test_add_service_sms_sender_switches_default(client, notify_db_session):
def test_add_service_sms_sender_return_404_when_service_does_not_exist(client):
data = {"sms_sender": "12345", "is_default": False}
response = client.post(
"/service/{}/sms-sender".format(uuid.uuid4()),
f"/service/{uuid.uuid4()}/sms-sender",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3164,7 +3181,7 @@ def test_update_service_sms_sender(client, notify_db_session):
"is_default": False,
}
response = client.post(
"/service/{}/sms-sender/{}".format(service.id, service_sms_sender.id),
f"/service/{service.id}/sms-sender/{service_sms_sender.id}",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3188,7 +3205,7 @@ def test_update_service_sms_sender_switches_default(client, notify_db_session):
"is_default": True,
}
response = client.post(
"/service/{}/sms-sender/{}".format(service.id, service_sms_sender.id),
f"/service/{service.id}/sms-sender/{service_sms_sender.id}",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3221,7 +3238,7 @@ def test_update_service_sms_sender_does_not_allow_sender_update_for_inbound_numb
"inbound_number_id": str(inbound_number.id),
}
response = client.post(
"/service/{}/sms-sender/{}".format(service.id, service_sms_sender.id),
f"/service/{service.id}/sms-sender/{service_sms_sender.id}",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3234,7 +3251,7 @@ def test_update_service_sms_sender_does_not_allow_sender_update_for_inbound_numb
def test_update_service_sms_sender_return_404_when_service_does_not_exist(client):
data = {"sms_sender": "12345", "is_default": False}
response = client.post(
"/service/{}/sms-sender/{}".format(uuid.uuid4(), uuid.uuid4()),
f"/service/{uuid.uuid4()}/sms-sender/{uuid.uuid4()}",
data=json.dumps(data),
headers=[
("Content-Type", "application/json"),
@@ -3288,9 +3305,7 @@ def test_get_service_sms_sender_by_id(client, notify_db_session):
service=create_service(), sms_sender="1235", is_default=False
)
response = client.get(
"/service/{}/sms-sender/{}".format(
service_sms_sender.service_id, service_sms_sender.id
),
f"/service/{service_sms_sender.service_id}/sms-sender/{service_sms_sender.id}",
headers=[
("Content-Type", "application/json"),
create_admin_authorization_header(),
@@ -3307,7 +3322,7 @@ def test_get_service_sms_sender_by_id_returns_404_when_service_does_not_exist(
service=create_service(), sms_sender="1235", is_default=False
)
response = client.get(
"/service/{}/sms-sender/{}".format(uuid.uuid4(), service_sms_sender.id),
f"/service/{uuid.uuid4()}/sms-sender/{service_sms_sender.id}",
headers=[
("Content-Type", "application/json"),
create_admin_authorization_header(),
@@ -3321,7 +3336,7 @@ def test_get_service_sms_sender_by_id_returns_404_when_sms_sender_does_not_exist
):
service = create_service()
response = client.get(
"/service/{}/sms-sender/{}".format(service.id, uuid.uuid4()),
f"/service/{service.id}/sms-sender/{uuid.uuid4()}",
headers=[
("Content-Type", "application/json"),
create_admin_authorization_header(),
@@ -3335,7 +3350,7 @@ def test_get_service_sms_senders_for_service(client, notify_db_session):
service=create_service(), sms_sender="second", is_default=False
)
response = client.get(
"/service/{}/sms-sender".format(service_sms_sender.service_id),
f"/service/{service_sms_sender.service_id}/sms-sender",
headers=[
("Content-Type", "application/json"),
create_admin_authorization_header(),
@@ -3354,7 +3369,7 @@ def test_get_service_sms_senders_for_service_returns_empty_list_when_service_doe
client,
):
response = client.get(
"/service/{}/sms-sender".format(uuid.uuid4()),
f"/service/{uuid.uuid4()}/sms-sender",
headers=[
("Content-Type", "application/json"),
create_admin_authorization_header(),
@@ -3386,15 +3401,15 @@ def test_get_organization_for_service_id_return_empty_dict_if_service_not_in_org
def test_get_monthly_notification_data_by_service(sample_service, admin_request):
create_ft_notification_status(
date(2019, 4, 17),
notification_type="sms",
notification_type=NotificationType.SMS,
service=sample_service,
notification_status="delivered",
notification_status=NotificationStatus.DELIVERED,
)
create_ft_notification_status(
date(2019, 3, 5),
notification_type="email",
notification_type=NotificationType.EMAIL,
service=sample_service,
notification_status="sending",
notification_status=NotificationStatus.SENDING,
count=4,
)
response = admin_request.get(
@@ -3408,7 +3423,7 @@ def test_get_monthly_notification_data_by_service(sample_service, admin_request)
"2019-03-01",
str(sample_service.id),
"Sample service",
"email",
NotificationType.EMAIL,
4,
0,
0,
@@ -3420,7 +3435,7 @@ def test_get_monthly_notification_data_by_service(sample_service, admin_request)
"2019-04-01",
str(sample_service.id),
"Sample service",
"sms",
NotificationType.SMS,
0,
1,
0,