mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 16:31:15 -05:00
35b20ba363
Last year we had an issue with the daily limit cache and the query that was populating it. As a result we have not been checking the daily limit properly. This PR should correct all that. The daily limit cache is not being incremented in app.notifications.process_notifications.persist_notification, this method is and should always be the only method used to create a notification. We increment the daily limit cache is redis is enabled (and it is always enabled for production) and the key type for the notification is team or normal. We check if the daily limit is exceed in many places: - app.celery.tasks.process_job - app.v2.notifications.post_notifications.post_notification - app.v2.notifications.post_notifications.post_precompiled_letter_notification - app.service.send_notification.send_one_off_notification - app.service.send_notification.send_pdf_letter_notification If the daily limits cache is not found, set the cache to 0 with an expiry of 24 hours. The daily limit cache key is service_id-yyy-mm-dd-count, so each day a new cache is created. The best thing about this PR is that the app.service_dao.fetch_todays_total_message_count query has been removed. This query was not performant and had been wrong for ages.
697 lines
31 KiB
Python
697 lines
31 KiB
Python
from datetime import datetime
|
||
|
||
import pytest
|
||
from flask import current_app
|
||
from freezegun import freeze_time
|
||
from notifications_utils import SMS_CHAR_COUNT_LIMIT
|
||
|
||
import app
|
||
from app.dao import templates_dao
|
||
from app.models import EMAIL_TYPE, INTERNATIONAL_LETTERS, LETTER_TYPE, SMS_TYPE
|
||
from app.notifications.process_notifications import (
|
||
create_content_for_notification,
|
||
)
|
||
from app.notifications.validators import (
|
||
check_if_service_can_send_files_by_email,
|
||
check_is_message_too_long,
|
||
check_notification_content_is_not_empty,
|
||
check_rate_limiting,
|
||
check_reply_to,
|
||
check_service_email_reply_to_id,
|
||
check_service_letter_contact_id,
|
||
check_service_over_api_rate_limit,
|
||
check_service_over_daily_message_limit,
|
||
check_service_sms_sender_id,
|
||
check_template_is_active,
|
||
check_template_is_for_notification_type,
|
||
get_service_daily_limit_cache_value,
|
||
service_can_send_to_recipient,
|
||
validate_address,
|
||
validate_and_format_recipient,
|
||
validate_template,
|
||
)
|
||
from app.serialised_models import (
|
||
SerialisedAPIKeyCollection,
|
||
SerialisedService,
|
||
SerialisedTemplate,
|
||
)
|
||
from app.utils import get_template_instance
|
||
from app.v2.errors import BadRequestError, RateLimitError, TooManyRequestsError
|
||
from tests.app.db import (
|
||
create_api_key,
|
||
create_letter_contact,
|
||
create_reply_to_email,
|
||
create_service,
|
||
create_service_guest_list,
|
||
create_service_sms_sender,
|
||
create_template,
|
||
)
|
||
from tests.conftest import set_config
|
||
|
||
|
||
# all of these tests should have redis enabled (except where we specifically disable it)
|
||
@pytest.fixture(scope='module', autouse=True)
|
||
def enable_redis(notify_api):
|
||
with set_config(notify_api, 'REDIS_ENABLED', True):
|
||
yield
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['team', 'normal'])
|
||
def test_check_service_message_limit_in_cache_under_message_limit_passes(
|
||
key_type,
|
||
sample_service,
|
||
mocker):
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
mock_get = mocker.patch('app.notifications.validators.redis_store.get', return_value=1)
|
||
mock_set = mocker.patch('app.notifications.validators.redis_store.set')
|
||
check_service_over_daily_message_limit(key_type, serialised_service)
|
||
mock_get.assert_called_once_with(f'{serialised_service.id}-{datetime.utcnow().strftime("%Y-%m-%d")}-count')
|
||
assert mock_get.called
|
||
assert not mock_set.called
|
||
|
||
|
||
def test_should_not_interact_with_cache_for_test_key(sample_service, mocker):
|
||
mocker.patch('app.notifications.validators.redis_store')
|
||
mock_get = mocker.patch('app.notifications.validators.redis_store.get', side_effect=[None])
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
check_service_over_daily_message_limit('test', serialised_service)
|
||
assert not app.notifications.validators.redis_store.mock_calls
|
||
assert not mock_get.called
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['team', 'normal'])
|
||
def test_should_set_cache_value_as_zero_if_cache_not_set(
|
||
key_type,
|
||
sample_template,
|
||
sample_service,
|
||
mocker
|
||
):
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
with freeze_time("2016-01-01 12:00:00.000000"):
|
||
mocker.patch('app.notifications.validators.redis_store.get', return_value=None)
|
||
mocker.patch('app.notifications.validators.redis_store.set')
|
||
check_service_over_daily_message_limit(key_type, serialised_service)
|
||
app.notifications.validators.redis_store.set.assert_called_with(
|
||
str(sample_service.id) + "-2016-01-01-count", 0, ex=86400
|
||
)
|
||
|
||
|
||
def test_does_nothing_if_redis_disabled(notify_api, sample_service, mocker):
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
with set_config(notify_api, 'REDIS_ENABLED', False):
|
||
mock_cache_key = mocker.patch('notifications_utils.clients.redis.daily_limit_cache_key')
|
||
check_service_over_daily_message_limit('normal', serialised_service)
|
||
|
||
assert mock_cache_key.method_calls == []
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['team', 'normal'])
|
||
def test_check_service_message_limit_over_message_limit_fails(key_type, mocker, notify_db_session):
|
||
with freeze_time("2016-01-01 12:00:00.000000"):
|
||
service = create_service(restricted=True, message_limit=4)
|
||
mocker.patch('app.redis_store.get', return_value=5)
|
||
|
||
with pytest.raises(TooManyRequestsError) as e:
|
||
check_service_over_daily_message_limit(key_type, service)
|
||
assert e.value.status_code == 429
|
||
assert e.value.message == 'Exceeded send limits (4) for today'
|
||
assert e.value.fields == []
|
||
|
||
|
||
@pytest.mark.parametrize('template_type, notification_type',
|
||
[(EMAIL_TYPE, EMAIL_TYPE),
|
||
(SMS_TYPE, SMS_TYPE)])
|
||
def test_check_template_is_for_notification_type_pass(template_type, notification_type):
|
||
assert check_template_is_for_notification_type(notification_type=notification_type,
|
||
template_type=template_type) is None
|
||
|
||
|
||
@pytest.mark.parametrize('template_type, notification_type',
|
||
[(SMS_TYPE, EMAIL_TYPE),
|
||
(EMAIL_TYPE, SMS_TYPE)])
|
||
def test_check_template_is_for_notification_type_fails_when_template_type_does_not_match_notification_type(
|
||
template_type, notification_type):
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_template_is_for_notification_type(notification_type=notification_type,
|
||
template_type=template_type)
|
||
assert e.value.status_code == 400
|
||
error_message = '{0} template is not suitable for {1} notification'.format(template_type, notification_type)
|
||
assert e.value.message == error_message
|
||
assert e.value.fields == [{'template': error_message}]
|
||
|
||
|
||
def test_check_template_is_active_passes(sample_template):
|
||
assert check_template_is_active(sample_template) is None
|
||
|
||
|
||
def test_check_template_is_active_fails(sample_template):
|
||
sample_template.archived = True
|
||
from app.dao.templates_dao import dao_update_template
|
||
dao_update_template(sample_template)
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_template_is_active(sample_template)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'Template has been deleted'
|
||
assert e.value.fields == [{'template': 'Template has been deleted'}]
|
||
|
||
|
||
@pytest.mark.parametrize('key_type',
|
||
['test', '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)
|
||
assert service_can_send_to_recipient(trial_mode_service.users[0].email_address,
|
||
key_type,
|
||
serialised_service) is None
|
||
assert service_can_send_to_recipient(trial_mode_service.users[0].mobile_number,
|
||
key_type,
|
||
serialised_service) is None
|
||
|
||
|
||
@pytest.mark.parametrize('key_type',
|
||
['test', 'normal'])
|
||
def test_service_can_send_to_recipient_passes_for_live_service_non_team_member(key_type, sample_service):
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
assert service_can_send_to_recipient("some_other_email@test.com",
|
||
key_type,
|
||
serialised_service) is None
|
||
assert service_can_send_to_recipient('07513332413',
|
||
key_type,
|
||
serialised_service) is None
|
||
|
||
|
||
def test_service_can_send_to_recipient_passes_for_guest_list_recipient_passes(sample_service):
|
||
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) is None
|
||
create_service_guest_list(sample_service, mobile_number='07513332413')
|
||
assert service_can_send_to_recipient('07513332413',
|
||
'team',
|
||
sample_service) is None
|
||
|
||
|
||
@pytest.mark.parametrize('recipient', [
|
||
{"email_address": "some_other_email@test.com"},
|
||
{"mobile_number": "07513332413"},
|
||
])
|
||
def test_service_can_send_to_recipient_fails_when_ignoring_guest_list(
|
||
notify_db,
|
||
notify_db_session,
|
||
sample_service,
|
||
recipient,
|
||
):
|
||
create_service_guest_list(sample_service, **recipient)
|
||
with pytest.raises(BadRequestError) as exec_info:
|
||
service_can_send_to_recipient(
|
||
next(iter(recipient.values())),
|
||
'team',
|
||
sample_service,
|
||
allow_guest_list_recipients=False,
|
||
)
|
||
assert exec_info.value.status_code == 400
|
||
assert exec_info.value.message == 'Can’t send to this recipient using a team-only API key'
|
||
assert exec_info.value.fields == []
|
||
|
||
|
||
@pytest.mark.parametrize('recipient', ['07513332413', 'some_other_email@test.com'])
|
||
@pytest.mark.parametrize('key_type, error_message',
|
||
[('team', 'Can’t send to this recipient using a team-only API key'),
|
||
('normal',
|
||
"Can’t send to this recipient when service is in trial mode – see https://www.notifications.service.gov.uk/trial-mode")]) # noqa
|
||
def test_service_can_send_to_recipient_fails_when_recipient_is_not_on_team(
|
||
recipient,
|
||
key_type,
|
||
error_message,
|
||
notify_db_session,
|
||
):
|
||
trial_mode_service = create_service(service_name='trial mode', restricted=True)
|
||
with pytest.raises(BadRequestError) as exec_info:
|
||
service_can_send_to_recipient(recipient,
|
||
key_type,
|
||
trial_mode_service)
|
||
assert exec_info.value.status_code == 400
|
||
assert exec_info.value.message == error_message
|
||
assert exec_info.value.fields == []
|
||
|
||
|
||
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)
|
||
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 == []
|
||
|
||
|
||
@pytest.mark.parametrize('char_count', [612, 0, 494, 200, 918])
|
||
@pytest.mark.parametrize('show_prefix', [True, False])
|
||
@pytest.mark.parametrize('template_type', ['sms', 'email', 'letter'])
|
||
def test_check_is_message_too_long_passes(notify_db_session, show_prefix, char_count, template_type):
|
||
service = create_service(prefix_sms=show_prefix)
|
||
t = create_template(service=service, content='a' * char_count, template_type=template_type)
|
||
template = templates_dao.dao_get_template_by_id_and_service_id(template_id=t.id, service_id=service.id)
|
||
template_with_content = get_template_instance(template=template.__dict__, values={})
|
||
assert check_is_message_too_long(template_with_content) is None
|
||
|
||
|
||
@pytest.mark.parametrize('char_count', [919, 6000])
|
||
@pytest.mark.parametrize('show_prefix', [True, False])
|
||
def test_check_is_message_too_long_fails(notify_db_session, show_prefix, char_count):
|
||
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')
|
||
template = templates_dao.dao_get_template_by_id_and_service_id(template_id=t.id, service_id=service.id)
|
||
template_with_content = get_template_instance(template=template.__dict__, values={})
|
||
check_is_message_too_long(template_with_content)
|
||
assert e.value.status_code == 400
|
||
expected_message = f'Your message is too long. '\
|
||
f'Text messages cannot be longer than {SMS_CHAR_COUNT_LIMIT} characters. '\
|
||
f'Your message is {char_count} characters long.'
|
||
assert e.value.message == expected_message
|
||
assert e.value.fields == []
|
||
|
||
|
||
def test_check_is_message_too_long_passes_for_long_email(sample_service):
|
||
email_character_count = 2_000_001
|
||
t = create_template(service=sample_service, content='a' * email_character_count, template_type='email')
|
||
template = templates_dao.dao_get_template_by_id_and_service_id(template_id=t.id,
|
||
service_id=t.service_id)
|
||
template_with_content = get_template_instance(template=template.__dict__, values={})
|
||
template_with_content.values
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_is_message_too_long(template_with_content)
|
||
assert e.value.status_code == 400
|
||
expected_message = (
|
||
'Your message is too long. ' +
|
||
'Emails cannot be longer than 2000000 bytes. ' +
|
||
'Your message is 2000001 bytes.'
|
||
)
|
||
assert e.value.message == expected_message
|
||
assert e.value.fields == []
|
||
|
||
|
||
def test_check_notification_content_is_not_empty_passes(notify_api, mocker, sample_service):
|
||
template_id = create_template(sample_service, content="Content is not empty").id
|
||
template = SerialisedTemplate.from_id_and_service_id(
|
||
template_id=template_id,
|
||
service_id=sample_service.id
|
||
)
|
||
template_with_content = create_content_for_notification(template, {})
|
||
assert check_notification_content_is_not_empty(template_with_content) is None
|
||
|
||
|
||
@pytest.mark.parametrize('template_content,notification_values', [
|
||
("", {}),
|
||
("((placeholder))", {"placeholder": ""})
|
||
])
|
||
def test_check_notification_content_is_not_empty_fails(
|
||
notify_api, mocker, sample_service, template_content, notification_values
|
||
):
|
||
template_id = create_template(sample_service, content=template_content).id
|
||
template = SerialisedTemplate.from_id_and_service_id(
|
||
template_id=template_id,
|
||
service_id=sample_service.id
|
||
)
|
||
template_with_content = create_content_for_notification(template, notification_values)
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_notification_content_is_not_empty(template_with_content)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'Your message is empty.'
|
||
assert e.value.fields == []
|
||
|
||
|
||
def test_validate_template(sample_service):
|
||
template = create_template(sample_service, template_type="email")
|
||
validate_template(template.id, {}, sample_service, "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")
|
||
mock_check_type = mocker.patch('app.notifications.validators.check_template_is_for_notification_type')
|
||
mock_check_if_active = mocker.patch('app.notifications.validators.check_template_is_active')
|
||
mock_create_conent = mocker.patch(
|
||
'app.notifications.validators.create_content_for_notification', return_value="content"
|
||
)
|
||
mock_check_not_empty = mocker.patch('app.notifications.validators.check_notification_content_is_not_empty')
|
||
mock_check_message_is_too_long = mocker.patch('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
|
||
)
|
||
|
||
mock_check_type.assert_called_once_with("email", "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")
|
||
if check_char_count:
|
||
mock_check_message_is_too_long.assert_called_once_with("content")
|
||
else:
|
||
assert not mock_check_message_is_too_long.called
|
||
|
||
|
||
def test_validate_template_calls_all_validators_exception_message_too_long(mocker, fake_uuid, sample_service):
|
||
template = create_template(sample_service, template_type="email")
|
||
mock_check_type = mocker.patch('app.notifications.validators.check_template_is_for_notification_type')
|
||
mock_check_if_active = mocker.patch('app.notifications.validators.check_template_is_active')
|
||
mock_create_conent = mocker.patch(
|
||
'app.notifications.validators.create_content_for_notification', return_value="content"
|
||
)
|
||
mock_check_not_empty = mocker.patch('app.notifications.validators.check_notification_content_is_not_empty')
|
||
mock_check_message_is_too_long = mocker.patch('app.notifications.validators.check_is_message_too_long')
|
||
template, template_with_content = validate_template(template.id, {}, sample_service, "email",
|
||
check_char_count=False)
|
||
|
||
mock_check_type.assert_called_once_with("email", "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")
|
||
assert not mock_check_message_is_too_long.called
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['team', 'live', 'test'])
|
||
def test_that_when_exceed_rate_limit_request_fails(
|
||
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)
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
serialised_api_key = SerialisedAPIKeyCollection.from_service_id(serialised_service.id)[0]
|
||
|
||
with pytest.raises(RateLimitError) as e:
|
||
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
|
||
|
||
assert app.redis_store.exceeded_rate_limit.called_with(
|
||
"{}-{}".format(str(sample_service.id), api_key.key_type),
|
||
sample_service.rate_limit,
|
||
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.fields == []
|
||
|
||
|
||
def test_that_when_not_exceeded_rate_limit_request_succeeds(
|
||
sample_service,
|
||
mocker):
|
||
with freeze_time("2016-01-01 12:00:00.000000"):
|
||
mocker.patch('app.redis_store.exceeded_rate_limit', return_value=False)
|
||
|
||
sample_service.restricted = True
|
||
api_key = create_api_key(sample_service)
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
serialised_api_key = SerialisedAPIKeyCollection.from_service_id(serialised_service.id)[0]
|
||
|
||
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
|
||
assert app.redis_store.exceeded_rate_limit.called_with(
|
||
"{}-{}".format(str(sample_service.id), api_key.key_type),
|
||
3000,
|
||
60
|
||
)
|
||
|
||
|
||
def test_should_not_rate_limit_if_limiting_is_disabled(
|
||
sample_service,
|
||
mocker):
|
||
with freeze_time("2016-01-01 12:00:00.000000"):
|
||
current_app.config['API_RATE_LIMIT_ENABLED'] = False
|
||
|
||
mocker.patch('app.redis_store.exceeded_rate_limit', return_value=False)
|
||
|
||
sample_service.restricted = True
|
||
create_api_key(sample_service)
|
||
serialised_service = SerialisedService.from_id(sample_service.id)
|
||
serialised_api_key = SerialisedAPIKeyCollection.from_service_id(serialised_service.id)[0]
|
||
|
||
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
|
||
assert not app.redis_store.exceeded_rate_limit.called
|
||
|
||
|
||
def test_check_rate_limiting_checks_daily_limit_if_trial_mode_service(
|
||
notify_db_session, mocker
|
||
):
|
||
mock_daily_limit = mocker.patch('app.notifications.validators.check_service_over_daily_message_limit')
|
||
trial_mode_service = create_service(restricted=True, message_limit=5)
|
||
team_key = create_api_key(service=trial_mode_service, key_type='team')
|
||
check_rate_limiting(trial_mode_service, team_key)
|
||
mock_daily_limit.assert_called_once_with('team', trial_mode_service)
|
||
|
||
|
||
def test_check_rate_limiting_checks_daily_limit_if_live_service(
|
||
notify_db_session, mocker
|
||
):
|
||
mock_daily_limit = mocker.patch('app.notifications.validators.check_service_over_daily_message_limit')
|
||
live_service = create_service(restricted=False, message_limit=5)
|
||
live_key = create_api_key(service=live_service, key_type='normal')
|
||
check_rate_limiting(service=live_service, api_key=live_key)
|
||
mock_daily_limit.assert_called_once_with('normal', live_service)
|
||
|
||
|
||
def test_check_rate_limiting_checks_rate_limit_if_trial_mode_service(
|
||
notify_db_session, mocker
|
||
):
|
||
mock_rate_limit = mocker.patch('app.notifications.validators.check_service_over_api_rate_limit')
|
||
trial_mode_service = create_service(restricted=True, message_limit=5)
|
||
team_key = create_api_key(service=trial_mode_service, key_type='team')
|
||
check_rate_limiting(trial_mode_service, team_key)
|
||
mock_rate_limit.assert_called_once_with(trial_mode_service, team_key)
|
||
|
||
|
||
def test_check_rate_limiting_checks_rate_limit_if_live_service(
|
||
notify_db_session, mocker
|
||
):
|
||
mock_rate_limit = mocker.patch('app.notifications.validators.check_service_over_api_rate_limit')
|
||
live_service = create_service(restricted=False, message_limit=5)
|
||
live_key = create_api_key(service=live_service, key_type='normal')
|
||
check_rate_limiting(service=live_service, api_key=live_key)
|
||
mock_rate_limit.assert_called_once_with(live_service, live_key)
|
||
|
||
|
||
def test_get_service_daily_limit_cache_value(
|
||
notify_api, notify_db_session, mocker
|
||
):
|
||
mock_get = mocker.patch('app.redis_store.get', return_value=5)
|
||
service = create_service()
|
||
with set_config(notify_api, 'REDIS_ENABLED', True):
|
||
get_service_daily_limit_cache_value('normal', service)
|
||
mock_get.assert_called_once_with(f'{str(service.id)}-{datetime.utcnow().strftime("%Y-%m-%d")}-count')
|
||
|
||
|
||
def test_get_service_daily_limit_cache_value_return_zero_and_sets_cache_if_key_not_found_in_cache(
|
||
notify_api, notify_db_session, mocker
|
||
):
|
||
mock_get = mocker.patch('app.redis_store.get', return_value=None)
|
||
mock_set = mocker.patch('app.redis_store.set')
|
||
service = create_service()
|
||
with set_config(notify_api, 'REDIS_ENABLED', True):
|
||
assert get_service_daily_limit_cache_value('normal', service) == 0
|
||
mock_get.assert_called_once_with(f'{str(service.id)}-{datetime.utcnow().strftime("%Y-%m-%d")}-count')
|
||
mock_set.assert_called_once_with(
|
||
f'{str(service.id)}-{datetime.utcnow().strftime("%Y-%m-%d")}-count',
|
||
0,
|
||
ex=86400
|
||
)
|
||
|
||
|
||
def test_get_service_daily_limit_cache_value_returns_zero_if_test_api_key(
|
||
notify_api, notify_db_session, mocker
|
||
):
|
||
mock_get = mocker.patch('app.redis_store.get', return_value=None)
|
||
mock_set = mocker.patch('app.redis_store.set')
|
||
service = create_service()
|
||
with set_config(notify_api, 'REDIS_ENABLED', True):
|
||
assert get_service_daily_limit_cache_value('test', service) == 0
|
||
assert not mock_get.called
|
||
assert not mock_set.called
|
||
|
||
|
||
def test_get_service_daily_limit_cache_value_returns_zero_if_redis_is_not_enabled(
|
||
notify_api, notify_db_session, mocker
|
||
):
|
||
mock_get = mocker.patch('app.redis_store.get', return_value=None)
|
||
mock_set = mocker.patch('app.redis_store.set')
|
||
service = create_service()
|
||
# redis is not enabled for the test environment but is always enabled for production
|
||
assert get_service_daily_limit_cache_value('test', service) == 0
|
||
assert not mock_get.called
|
||
assert not mock_set.called
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['test', 'normal'])
|
||
def test_rejects_api_calls_with_international_numbers_if_service_does_not_allow_int_sms(
|
||
key_type,
|
||
notify_db_session,
|
||
):
|
||
service = create_service(service_permissions=[SMS_TYPE])
|
||
service_model = SerialisedService.from_id(service.id)
|
||
with pytest.raises(BadRequestError) as e:
|
||
validate_and_format_recipient('20-12-1234-1234', key_type, service_model, SMS_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'Cannot send to international mobile numbers'
|
||
assert e.value.fields == []
|
||
|
||
|
||
@pytest.mark.parametrize('key_type', ['test', 'normal'])
|
||
def test_allows_api_calls_with_international_numbers_if_service_does_allow_int_sms(
|
||
key_type, sample_service_full_permissions):
|
||
service_model = SerialisedService.from_id(sample_service_full_permissions.id)
|
||
result = validate_and_format_recipient('20-12-1234-1234', key_type, service_model, SMS_TYPE)
|
||
assert result == '201212341234'
|
||
|
||
|
||
def test_rejects_api_calls_with_no_recipient():
|
||
with pytest.raises(BadRequestError) as e:
|
||
validate_and_format_recipient(None, 'key_type', 'service', 'SMS_TYPE')
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == "Recipient can't be empty"
|
||
|
||
|
||
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
|
||
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
|
||
|
||
|
||
def test_check_service_email_reply_to_where_email_reply_to_is_found(sample_service):
|
||
reply_to_address = create_reply_to_email(sample_service, "test@test.com")
|
||
assert check_service_email_reply_to_id(sample_service.id, reply_to_address.id, EMAIL_TYPE) == "test@test.com"
|
||
|
||
|
||
def test_check_service_email_reply_to_id_where_service_id_is_not_found(sample_service, fake_uuid):
|
||
reply_to_address = create_reply_to_email(sample_service, "test@test.com")
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_email_reply_to_id(fake_uuid, reply_to_address.id, EMAIL_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}' \
|
||
.format(reply_to_address.id, fake_uuid)
|
||
|
||
|
||
def test_check_service_email_reply_to_id_where_reply_to_id_is_not_found(sample_service, fake_uuid):
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_email_reply_to_id(sample_service.id, fake_uuid, EMAIL_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}' \
|
||
.format(fake_uuid, sample_service.id)
|
||
|
||
|
||
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
|
||
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
|
||
|
||
|
||
def test_check_service_sms_sender_id_where_sms_sender_id_is_found(sample_service):
|
||
sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
|
||
assert check_service_sms_sender_id(sample_service.id, sms_sender.id, SMS_TYPE) == '123456'
|
||
|
||
|
||
def test_check_service_sms_sender_id_where_service_id_is_not_found(sample_service, fake_uuid):
|
||
sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_sms_sender_id(fake_uuid, sms_sender.id, SMS_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'sms_sender_id {} does not exist in database for service id {}' \
|
||
.format(sms_sender.id, fake_uuid)
|
||
|
||
|
||
def test_check_service_sms_sender_id_where_sms_sender_is_not_found(sample_service, fake_uuid):
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_sms_sender_id(sample_service.id, fake_uuid, SMS_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'sms_sender_id {} does not exist in database for service id {}' \
|
||
.format(fake_uuid, sample_service.id)
|
||
|
||
|
||
def test_check_service_letter_contact_id_where_letter_contact_id_is_none():
|
||
assert check_service_letter_contact_id(None, None, 'letter') is None
|
||
|
||
|
||
def test_check_service_letter_contact_id_where_letter_contact_id_is_found(sample_service):
|
||
letter_contact = create_letter_contact(service=sample_service, contact_block='123456')
|
||
assert check_service_letter_contact_id(sample_service.id, letter_contact.id, LETTER_TYPE) == '123456'
|
||
|
||
|
||
def test_check_service_letter_contact_id_where_service_id_is_not_found(sample_service, fake_uuid):
|
||
letter_contact = create_letter_contact(service=sample_service, contact_block='123456')
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_letter_contact_id(fake_uuid, letter_contact.id, LETTER_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'letter_contact_id {} does not exist in database for service id {}' \
|
||
.format(letter_contact.id, fake_uuid)
|
||
|
||
|
||
def test_check_service_letter_contact_id_where_letter_contact_is_not_found(sample_service, fake_uuid):
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_service_letter_contact_id(sample_service.id, fake_uuid, LETTER_TYPE)
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == 'letter_contact_id {} does not exist in database for service id {}' \
|
||
.format(fake_uuid, sample_service.id)
|
||
|
||
|
||
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
|
||
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
|
||
|
||
|
||
def test_check_reply_to_email_type(sample_service):
|
||
reply_to_address = create_reply_to_email(sample_service, "test@test.com")
|
||
assert check_reply_to(sample_service.id, reply_to_address.id, EMAIL_TYPE) == 'test@test.com'
|
||
|
||
|
||
def test_check_reply_to_sms_type(sample_service):
|
||
sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
|
||
assert check_reply_to(sample_service.id, sms_sender.id, SMS_TYPE) == '123456'
|
||
|
||
|
||
def test_check_reply_to_letter_type(sample_service):
|
||
letter_contact = create_letter_contact(service=sample_service, contact_block='123456')
|
||
assert check_reply_to(sample_service.id, letter_contact.id, LETTER_TYPE) == '123456'
|
||
|
||
|
||
def test_check_if_service_can_send_files_by_email_raises_if_no_contact_link_set(sample_service):
|
||
with pytest.raises(BadRequestError) as e:
|
||
check_if_service_can_send_files_by_email(
|
||
service_contact_link=sample_service.contact_link,
|
||
service_id=sample_service.id
|
||
)
|
||
|
||
message = f"Send files by email has not been set up - add contact details for your service at " \
|
||
f"http://localhost:6012/services/{sample_service.id}/service-settings/send-files-by-email"
|
||
assert e.value.status_code == 400
|
||
assert e.value.message == message
|
||
|
||
|
||
def test_check_if_service_can_send_files_by_email_passes_if_contact_link_set(sample_service):
|
||
sample_service.contact_link = 'contact.me@gov.uk'
|
||
check_if_service_can_send_files_by_email(
|
||
service_contact_link=sample_service.contact_link,
|
||
service_id=sample_service.id
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize('key, address_line_3, expected_postage',
|
||
[('address_line_3', 'SW1 1AA', None),
|
||
('address_line_5', 'CANADA', 'rest-of-world'),
|
||
('address_line_3', 'GERMANY', 'europe')
|
||
])
|
||
def test_validate_address(notify_db_session, key, address_line_3, expected_postage):
|
||
service = create_service(service_permissions=[LETTER_TYPE, INTERNATIONAL_LETTERS])
|
||
data = {
|
||
'address_line_1': 'Prince Harry',
|
||
'address_line_2': 'Toronto',
|
||
key: address_line_3,
|
||
}
|
||
postage = validate_address(service, data)
|
||
assert postage == expected_postage
|