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

Merge branch 'main' into stvnrlly-paperless-api

Browse Source
This commit is contained in:
stvnrlly
2023-02-06 12:28:10 -05:00
parent 33bd3b3a31 dc06b411ca
commit 8b75424e29
87 changed files with 2484 additions and 2087 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
tests/app/notifications/test_notifications_sms_callbacks.py
View File

@@ -1,184 +0,0 @@
import pytest
from flask import json
from app.notifications.notifications_sms_callback import validate_callback_data
def firetext_post(client, data):
return client.post(
path='/notifications/sms/firetext',
data=data,
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
def mmg_post(client, data):
return client.post(
path='/notifications/sms/mmg',
data=data,
headers=[('Content-Type', 'application/json')])
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_should_not_need_auth(client, mocker):
mocker.patch('app.notifications.notifications_sms_callback.process_sms_client_response')
data = 'mobile=441234123123&status=0&reference=notification_id&time=2016-03-10 14:17:00'
response = firetext_post(client, data)
assert response.status_code == 200
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_should_return_400_if_empty_reference(client, mocker):
data = 'mobile=441234123123&status=0&reference=&time=2016-03-10 14:17:00'
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert json_resp['message'] == ['Firetext callback failed: reference missing']
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_should_return_400_if_no_reference(client, mocker):
data = 'mobile=441234123123&status=0&time=2016-03-10 14:17:00'
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert json_resp['message'] == ['Firetext callback failed: reference missing']
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_should_return_400_if_no_status(client, mocker):
data = 'mobile=441234123123&time=2016-03-10 14:17:00&reference=notification_id'
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert json_resp['message'] == ['Firetext callback failed: status missing']
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_should_return_200_and_call_task_with_valid_data(client, mocker):
mock_celery = mocker.patch(
'app.notifications.notifications_sms_callback.process_sms_client_response.apply_async')
data = 'mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference=notification_id'
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert json_resp['result'] == 'success'
mock_celery.assert_called_once_with(
['0', 'notification_id', 'Firetext', None],
queue='sms-callbacks',
)
@pytest.mark.skip(reason="Needs updating for TTS: Firetext removal")
def test_firetext_callback_including_a_code_should_return_200_and_call_task_with_valid_data(client, mocker):
mock_celery = mocker.patch(
'app.notifications.notifications_sms_callback.process_sms_client_response.apply_async')
data = 'mobile=441234123123&status=1&code=101&time=2016-03-10 14:17:00&reference=notification_id'
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert json_resp['result'] == 'success'
mock_celery.assert_called_once_with(
['1', 'notification_id', 'Firetext', '101'],
queue='sms-callbacks',
)
@pytest.mark.skip(reason="Needs updating for TTS: MMG removal")
def test_mmg_callback_should_not_need_auth(client, mocker, sample_notification):
mocker.patch('app.notifications.notifications_sms_callback.process_sms_client_response')
data = json.dumps({"reference": "mmg_reference",
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": "3",
"deliverytime": "2016-04-05 16:01:07"})
response = mmg_post(client, data)
assert response.status_code == 200
@pytest.mark.skip(reason="Needs updating for TTS: MMG removal")
def test_process_mmg_response_returns_400_for_malformed_data(client):
data = json.dumps({"reference": "mmg_reference",
"monkey": 'random thing',
"MSISDN": "447777349060",
"no_status": 00,
"deliverytime": "2016-04-05 16:01:07"})
response = mmg_post(client, data)
assert response.status_code == 400
json_data = json.loads(response.data)
assert json_data['result'] == 'error'
assert len(json_data['message']) == 2
assert "{} callback failed: {} missing".format('MMG', 'status') in json_data['message']
assert "{} callback failed: {} missing".format('MMG', 'CID') in json_data['message']
@pytest.mark.skip(reason="Needs updating for TTS: MMG removal")
def test_mmg_callback_should_return_200_and_call_task_with_valid_data(client, mocker):
mock_celery = mocker.patch(
'app.notifications.notifications_sms_callback.process_sms_client_response.apply_async')
data = json.dumps({"reference": "mmg_reference",
"CID": "notification_id",
"MSISDN": "447777349060",
"status": "3",
"substatus": "5",
"deliverytime": "2016-04-05 16:01:07"})
response = mmg_post(client, data)
assert response.status_code == 200
json_data = json.loads(response.data)
assert json_data['result'] == 'success'
mock_celery.assert_called_once_with(
['3', 'notification_id', 'MMG', '5'],
queue='sms-callbacks',
)
def test_validate_callback_data_returns_none_when_valid():
form = {'status': 'good',
'reference': 'send-sms-code'}
fields = ['status', 'reference']
client_name = 'sms client'
assert validate_callback_data(form, fields, client_name) is None
def test_validate_callback_data_return_errors_when_fields_are_empty():
form = {'monkey': 'good'}
fields = ['status', 'cid']
client_name = 'sms client'
errors = validate_callback_data(form, fields, client_name)
assert len(errors) == 2
assert "{} callback failed: {} missing".format(client_name, 'status') in errors
assert "{} callback failed: {} missing".format(client_name, 'cid') in errors
def test_validate_callback_data_can_handle_integers():
form = {'status': 00, 'cid': 'fsdfadfsdfas'}
fields = ['status', 'cid']
client_name = 'sms client'
result = validate_callback_data(form, fields, client_name)
assert result is None
def test_validate_callback_data_returns_error_for_empty_string():
form = {'status': '', 'cid': 'fsdfadfsdfas'}
fields = ['status', 'cid']
client_name = 'sms client'
result = validate_callback_data(form, fields, client_name)
assert result is not None
assert "{} callback failed: {} missing".format(client_name, 'status') in result

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

@@ -122,7 +122,7 @@ def test_persist_notification_with_optionals(sample_job, sample_api_key):
persist_notification(
template_id=sample_job.template.id,
template_version=sample_job.template.version,
recipient='+447111111111',
recipient='+12028675309',
service=sample_job.service,
personalisation=None,
notification_type='sms',
@@ -146,7 +146,7 @@ def test_persist_notification_with_optionals(sample_job, sample_api_key):
assert persisted_notification.client_reference == "ref from client"
assert persisted_notification.reference is None
assert persisted_notification.international is False
assert persisted_notification.phone_prefix == '44'
assert persisted_notification.phone_prefix == '1'
assert persisted_notification.rate_multiplier == 1
assert persisted_notification.created_by_id == sample_job.created_by_id
assert not persisted_notification.reply_to_text
@@ -295,15 +295,15 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(sample
@pytest.mark.parametrize("to_address, notification_type, expected", [
("+447700900000", "sms", True),
("+447700900111", "sms", True),
("+447700900222", "sms", True),
("07700900000", "sms", True),
("7700900111", "sms", True),
("+12028675000", "sms", True),
("+12028675111", "sms", True),
("+12028675222", "sms", True),
("2028675000", "sms", True),
("2028675111", "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),
("07515896969", "sms", False),
("2028675309", "sms", False),
("valid_email@test.com", "email", False)
])
def test_simulated_recipient(notify_api, to_address, notification_type, expected):
@@ -315,7 +315,7 @@ def test_simulated_recipient(notify_api, to_address, notification_type, expected
'simulate-delivered-2@notifications.service.gov.uk',
'simulate-delivered-2@notifications.service.gov.uk'
)
SIMULATED_SMS_NUMBERS = ('+447700900000', '+447700900111', '+447700900222')
SIMULATED_SMS_NUMBERS = ('+12028675000', '+12028675111', '+12028675222')
"""
formatted_address = None
@@ -330,12 +330,10 @@ def test_simulated_recipient(notify_api, to_address, notification_type, expected
@pytest.mark.parametrize('recipient, expected_international, expected_prefix, expected_units', [
('7900900123', False, '44', 1), # UK
('+447900900123', False, '44', 1), # UK
('07700910222', True, '44', 1), # UK (Jersey)
('07700900222', False, '44', 1), # TV number
('73122345678', True, '7', 1), # Russia
('360623400400', True, '36', 3)] # Hungary
('+447900900123', True, '44', 1), # UK
('+73122345678', True, '7', 1), # Russia
('+360623400400', True, '36', 1), # Hungary
('2028675309', False, '1', 1)] # USA
)
def test_persist_notification_with_international_info_stores_correct_info(
sample_job,
@@ -392,15 +390,12 @@ def test_persist_notification_with_international_info_does_not_store_for_email(
@pytest.mark.parametrize('recipient, expected_recipient_normalised', [
('7900900123', '447900900123'),
('+447900 900 123', '447900900123'),
(' 07700900222', '447700900222'),
('07700900222', '447700900222'),
(' 73122345678', '73122345678'),
('360623400400', '360623400400'),
('-077-00900222-', '447700900222'),
('(360623(400400)', '360623400400')
('+4407900900123', '+447900900123'),
('202-867-5309', '+12028675309'),
('1 202-867-5309', '+12028675309'),
('+1 (202) 867-5309', '+12028675309'),
('(202) 867-5309', '+12028675309'),
('2028675309', '+12028675309')
])
def test_persist_sms_notification_stores_normalised_number(
sample_job,

271
tests/app/notifications/test_receive_notification.py
View File

@@ -1,17 +1,13 @@
import base64
from base64 import b64encode
from datetime import datetime
import pytest
from flask import json
from freezegun import freeze_time
from app.models import EMAIL_TYPE, INBOUND_SMS_TYPE, SMS_TYPE, InboundSms
from app.notifications.receive_notifications import (
create_inbound_sms_object,
format_mmg_datetime,
format_mmg_message,
has_inbound_sms_permissions,
strip_leading_forty_four,
unescape_string,
)
from tests.app.db import (
@@ -22,57 +18,41 @@ from tests.app.db import (
from tests.conftest import set_config
def firetext_post(client, data, auth=True, password='testkey'):
headers = [
('Content-Type', 'application/x-www-form-urlencoded'),
]
if auth:
auth_value = base64.b64encode("notify:{}".format(password).encode('utf-8')).decode('utf-8')
headers.append(('Authorization', 'Basic ' + auth_value))
return client.post(
path='/notifications/sms/receive/firetext',
data=data,
headers=headers
)
def mmg_post(client, data, auth=True, password='testkey'):
def sns_post(client, data, auth=True, password='testkey'):
headers = [
('Content-Type', 'application/json'),
]
if auth:
auth_value = base64.b64encode("username:{}".format(password).encode('utf-8')).decode('utf-8')
headers.append(('Authorization', 'Basic ' + auth_value))
auth_value = b64encode(f"notify:{password}".encode())
headers.append(('Authorization', f"Basic {auth_value}"))
return client.post(
path='/notifications/sms/receive/mmg',
data=json.dumps(data),
path='/notifications/sms/receive/sns',
data={"Message": data},
headers=headers
)
def test_receive_notification_returns_received_to_mmg(client, mocker, sample_service_full_permissions):
@pytest.mark.skip(reason="Need to implement SNS tests. Body here mostly from MMG")
def test_receive_notification_returns_received_to_sns(client, mocker, sample_service_full_permissions):
mocked = mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
prom_counter_labels_mock = mocker.patch('app.notifications.receive_notifications.INBOUND_SMS_COUNTER.labels')
data = {
"ID": "1234",
"MSISDN": "447700900855",
"Message": "Some message to notify",
"Trigger": "Trigger?",
"Number": sample_service_full_permissions.get_inbound_number(),
"Channel": "SMS",
"DateRecieved": "2012-06-27 12:33:00"
"originationNumber": "+12028675309",
"destinationNumber": sample_service_full_permissions.get_inbound_number(),
"messageKeyword": "JOIN",
"messageBody": "EXAMPLE",
"inboundMessageId": "cae173d2-66b9-564c-8309-21f858e9fb84",
"previousPublishedMessageId": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
}
response = mmg_post(client, data)
response = sns_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert result['status'] == 'ok'
assert result['result'] == 'success'
prom_counter_labels_mock.assert_called_once_with("mmg")
prom_counter_labels_mock.assert_called_once_with("sns")
prom_counter_labels_mock.return_value.inc.assert_called_once_with()
inbound_sms_id = InboundSms.query.all()[0].id
@@ -84,7 +64,7 @@ def test_receive_notification_returns_received_to_mmg(client, mocker, sample_ser
[SMS_TYPE],
[INBOUND_SMS_TYPE],
])
def test_receive_notification_from_mmg_without_permissions_does_not_persist(
def test_receive_notification_from_sns_without_permissions_does_not_persist(
client,
mocker,
notify_db_session,
@@ -101,42 +81,17 @@ def test_receive_notification_from_mmg_without_permissions_does_not_persist(
"Channel": "SMS",
"DateRecieved": "2012-06-27 12:33:00"
}
response = mmg_post(client, data)
response = sns_post(client, data)
assert response.status_code == 200
assert response.get_data(as_text=True) == 'RECEIVED'
parsed_response = json.loads(response.get_data(as_text=True))
assert parsed_response['result'] == 'success'
assert InboundSms.query.count() == 0
assert mocked.called is False
@pytest.mark.parametrize('permissions', [
[SMS_TYPE],
[INBOUND_SMS_TYPE],
])
def test_receive_notification_from_firetext_without_permissions_does_not_persist(
client,
mocker,
notify_db_session,
permissions
):
service = create_service_with_inbound_number(inbound_number='07111111111', service_permissions=permissions)
mocker.patch("app.notifications.receive_notifications.dao_fetch_service_by_inbound_number",
return_value=service)
mocked_send_inbound_sms = mocker.patch(
"app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
mocker.patch("app.notifications.receive_notifications.has_inbound_sms_permissions", return_value=False)
data = "source=07999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
response = firetext_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert result['status'] == 'ok'
assert InboundSms.query.count() == 0
assert not mocked_send_inbound_sms.called
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
def test_receive_notification_without_permissions_does_not_create_inbound_even_with_inbound_number_set(
client, mocker, sample_service):
inbound_number = create_inbound_number('1', service_id=sample_service.id, active=True)
@@ -156,7 +111,7 @@ def test_receive_notification_without_permissions_does_not_create_inbound_even_w
"DateRecieved": "2012-06-27 12:33:00"
}
response = mmg_post(client, data)
response = sns_post(client, data)
assert response.status_code == 200
assert len(InboundSms.query.all()) == 0
@@ -174,17 +129,6 @@ def test_check_permissions_for_inbound_sms(notify_db_session, permissions, expec
assert has_inbound_sms_permissions(service.permissions) is expected_response
@pytest.mark.parametrize('message, expected_output', [
('abc', 'abc'),
('', ''),
('lots+of+words', 'lots of words'),
('%F0%9F%93%A9+%F0%9F%93%A9+%F0%9F%93%A9', '📩 📩 📩'),
('x+%2B+y', 'x + y')
])
def test_format_mmg_message(message, expected_output):
assert format_mmg_message(message) == expected_output
@pytest.mark.parametrize('raw, expected', [
(
'😬',
@@ -215,20 +159,8 @@ def test_unescape_string(raw, expected):
assert unescape_string(raw) == expected
@pytest.mark.parametrize('provider_date, expected_output', [
('2017-01-21+11%3A56%3A11', datetime(2017, 1, 21, 11, 56, 11)),
('2017-05-21+11%3A56%3A11', datetime(2017, 5, 21, 11, 56, 11))
])
def test_format_mmg_datetime(provider_date, expected_output):
assert format_mmg_datetime(provider_date) == expected_output
@freeze_time('2020-05-14 14:30:00')
def test_format_mmg_datetime_returns_now_if_cannot_parse_date():
assert format_mmg_datetime('13-05-2020 08%3A37%3A43') == datetime.utcnow()
def test_create_inbound_mmg_sms_object(sample_service_full_permissions):
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
def test_create_inbound_sns_sms_object(sample_service_full_permissions):
data = {
'Message': 'hello+there+%F0%9F%93%A9',
'Number': sample_service_full_permissions.get_inbound_number(),
@@ -237,8 +169,8 @@ def test_create_inbound_mmg_sms_object(sample_service_full_permissions):
'ID': 'bar',
}
inbound_sms = create_inbound_sms_object(sample_service_full_permissions, format_mmg_message(data["Message"]),
data["MSISDN"], data["ID"], data["DateRecieved"], "mmg")
inbound_sms = create_inbound_sms_object(sample_service_full_permissions, data["Message"],
data["MSISDN"], data["ID"], data["DateRecieved"], "sns")
assert inbound_sms.service_id == sample_service_full_permissions.id
assert inbound_sms.notify_number == sample_service_full_permissions.get_inbound_number()
@@ -247,10 +179,11 @@ def test_create_inbound_mmg_sms_object(sample_service_full_permissions):
assert inbound_sms.provider_reference == 'bar'
assert inbound_sms._content != 'hello there 📩'
assert inbound_sms.content == 'hello there 📩'
assert inbound_sms.provider == 'mmg'
assert inbound_sms.provider == 'sns'
def test_create_inbound_mmg_sms_object_uses_inbound_number_if_set(sample_service_full_permissions):
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
def test_create_inbound_sns_sms_object_uses_inbound_number_if_set(sample_service_full_permissions):
sample_service_full_permissions.sms_sender = 'foo'
inbound_number = sample_service_full_permissions.get_inbound_number()
@@ -264,17 +197,18 @@ def test_create_inbound_mmg_sms_object_uses_inbound_number_if_set(sample_service
inbound_sms = create_inbound_sms_object(
sample_service_full_permissions,
format_mmg_message(data["Message"]),
data["Message"],
data["MSISDN"],
data["ID"],
data["DateRecieved"],
"mmg"
"sns"
)
assert inbound_sms.service_id == sample_service_full_permissions.id
assert inbound_sms.notify_number == inbound_number
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
@pytest.mark.parametrize('notify_number', ['foo', 'baz'], ids=['two_matching_services', 'no_matching_services'])
def test_receive_notification_error_if_not_single_matching_service(client, notify_db_session, notify_number):
create_service_with_inbound_number(
@@ -295,7 +229,7 @@ def test_receive_notification_error_if_not_single_matching_service(client, notif
'DateRecieved': '2017-01-02 03:04:05',
'ID': 'bar',
}
response = mmg_post(client, data)
response = sns_post(client, data)
# we still return 'RECEIVED' to MMG
assert response.status_code == 200
@@ -303,107 +237,7 @@ def test_receive_notification_error_if_not_single_matching_service(client, notif
assert InboundSms.query.count() == 0
def test_receive_notification_returns_received_to_firetext(notify_db_session, client, mocker):
mocked = mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
prom_counter_labels_mock = mocker.patch('app.notifications.receive_notifications.INBOUND_SMS_COUNTER.labels')
service = create_service_with_inbound_number(
service_name='b', inbound_number='07111111111', service_permissions=[EMAIL_TYPE, SMS_TYPE, INBOUND_SMS_TYPE])
data = "source=07999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
response = firetext_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
prom_counter_labels_mock.assert_called_once_with("firetext")
prom_counter_labels_mock.return_value.inc.assert_called_once_with()
assert result['status'] == 'ok'
inbound_sms_id = InboundSms.query.all()[0].id
mocked.assert_called_once_with([str(inbound_sms_id), str(service.id)], queue="notify-internal-tasks")
def test_receive_notification_from_firetext_persists_message(notify_db_session, client, mocker):
mocked = mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
mocker.patch('app.notifications.receive_notifications.INBOUND_SMS_COUNTER')
service = create_service_with_inbound_number(
inbound_number='07111111111',
service_name='b',
service_permissions=[EMAIL_TYPE, SMS_TYPE, INBOUND_SMS_TYPE])
data = "source=07999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
response = firetext_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
persisted = InboundSms.query.first()
assert result['status'] == 'ok'
assert persisted.notify_number == '07111111111'
assert persisted.user_number == '447999999999'
assert persisted.service == service
assert persisted.content == 'this is a message'
assert persisted.provider == 'firetext'
assert persisted.provider_date == datetime(2017, 1, 1, 12, 0, 0, 0)
mocked.assert_called_once_with([str(persisted.id), str(service.id)], queue="notify-internal-tasks")
def test_receive_notification_from_firetext_persists_message_with_normalized_phone(notify_db_session, client, mocker):
mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
mocker.patch('app.notifications.receive_notifications.INBOUND_SMS_COUNTER')
create_service_with_inbound_number(
inbound_number='07111111111', service_name='b', service_permissions=[EMAIL_TYPE, SMS_TYPE, INBOUND_SMS_TYPE])
data = "source=(+44)7999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
response = firetext_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
persisted = InboundSms.query.first()
assert result['status'] == 'ok'
assert persisted.user_number == '447999999999'
def test_returns_ok_to_firetext_if_mismatched_sms_sender(notify_db_session, client, mocker):
mocked = mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
mocker.patch('app.notifications.receive_notifications.INBOUND_SMS_COUNTER')
create_service_with_inbound_number(
inbound_number='07111111199', service_name='b', service_permissions=[EMAIL_TYPE, SMS_TYPE, INBOUND_SMS_TYPE])
data = "source=(+44)7999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
response = firetext_post(client, data)
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert not InboundSms.query.all()
assert result['status'] == 'ok'
assert mocked.call_count == 0
@pytest.mark.parametrize(
'number, expected',
[
('447123123123', '07123123123'),
('447123123144', '07123123144'),
('07123123123', '07123123123'),
('447444444444', '07444444444')
]
)
def test_strip_leading_country_code(number, expected):
assert strip_leading_forty_four(number) == expected
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
@pytest.mark.parametrize("auth, keys, status_code", [
["testkey", ["testkey"], 200],
["", ["testkey"], 401],
@@ -414,31 +248,7 @@ def test_strip_leading_country_code(number, expected):
["", [], 401],
["testkey", [], 403],
])
def test_firetext_inbound_sms_auth(notify_db_session, notify_api, client, mocker, auth, keys, status_code):
mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
create_service_with_inbound_number(
service_name='b', inbound_number='07111111111', service_permissions=[EMAIL_TYPE, SMS_TYPE, INBOUND_SMS_TYPE]
)
data = "source=07999999999&destination=07111111111&message=this is a message&time=2017-01-01 12:00:00"
with set_config(notify_api, 'FIRETEXT_INBOUND_SMS_AUTH', keys):
response = firetext_post(client, data, auth=bool(auth), password=auth)
assert response.status_code == status_code
@pytest.mark.parametrize("auth, keys, status_code", [
["testkey", ["testkey"], 200],
["", ["testkey"], 401],
["wrong", ["testkey"], 403],
["testkey1", ["testkey1", "testkey2"], 200],
["testkey2", ["testkey1", "testkey2"], 200],
["wrong", ["testkey1", "testkey2"], 403],
["", [], 401],
["testkey", [], 403],
])
def test_mmg_inbound_sms_auth(notify_db_session, notify_api, client, mocker, auth, keys, status_code):
def test_sns_inbound_sms_auth(notify_db_session, notify_api, client, mocker, auth, keys, status_code):
mocker.patch("app.notifications.receive_notifications.tasks.send_inbound_sms_to_service.apply_async")
create_service_with_inbound_number(
@@ -456,10 +266,11 @@ def test_mmg_inbound_sms_auth(notify_db_session, notify_api, client, mocker, aut
}
with set_config(notify_api, 'MMG_INBOUND_SMS_AUTH', keys):
response = mmg_post(client, data, auth=bool(auth), password=auth)
response = sns_post(client, data, auth=bool(auth), password=auth)
assert response.status_code == status_code
@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
def test_create_inbound_sms_object_works_with_alphanumeric_sender(sample_service_full_permissions):
data = {
'Message': 'hello',
@@ -471,7 +282,7 @@ def test_create_inbound_sms_object_works_with_alphanumeric_sender(sample_service
inbound_sms = create_inbound_sms_object(
service=sample_service_full_permissions,
content=format_mmg_message(data["Message"]),
content=data["Message"],
from_number='ALPHANUM3R1C',
provider_ref='foo',
date_received=None,

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

@@ -167,10 +167,10 @@ def test_service_can_send_to_recipient_passes(key_type, notify_db_session):
@pytest.mark.parametrize('user_number, recipient_number', [
['0048601234567', '+486 012 34567'],
['07513332413', '(07513) 332413'],
['+12028675309', '202-867-5309'],
['+447513332413', '+44 (07513) 332413'],
])
def test_service_can_send_to_recipient_passes_with_non_normalised_number(sample_service, user_number, recipient_number):
def test_service_can_send_to_recipient_passes_with_non_normalized_number(sample_service, user_number, recipient_number):
sample_service.users[0].mobile_number = user_number
serialised_service = SerialisedService.from_id(sample_service.id)
@@ -181,7 +181,7 @@ def test_service_can_send_to_recipient_passes_with_non_normalised_number(sample_
@pytest.mark.parametrize('user_email, recipient_email', [
['test@example.com', 'TeSt@EXAMPLE.com'],
])
def test_service_can_send_to_recipient_passes_with_non_normalised_email(sample_service, user_email, recipient_email):
def test_service_can_send_to_recipient_passes_with_non_normalized_email(sample_service, user_email, recipient_email):
sample_service.users[0].email_address = user_email
serialised_service = SerialisedService.from_id(sample_service.id)
@@ -206,15 +206,15 @@ def test_service_can_send_to_recipient_passes_for_guest_list_recipient_passes(sa
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',
create_service_guest_list(sample_service, mobile_number='2028675309')
assert service_can_send_to_recipient('2028675309',
'team',
sample_service) is None
@pytest.mark.parametrize('recipient', [
{"email_address": "some_other_email@test.com"},
{"mobile_number": "07513332413"},
{"mobile_number": "2028675300"},
])
def test_service_can_send_to_recipient_fails_when_ignoring_guest_list(
notify_db_session,
@@ -234,7 +234,7 @@ def test_service_can_send_to_recipient_fails_when_ignoring_guest_list(
assert exec_info.value.fields == []
@pytest.mark.parametrize('recipient', ['07513332413', 'some_other_email@test.com'])
@pytest.mark.parametrize('recipient', ['2028675300', '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',
@@ -482,7 +482,7 @@ def test_validate_and_format_recipient_fails_when_international_number_and_servi
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)
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 == []
@@ -492,8 +492,8 @@ def test_validate_and_format_recipient_fails_when_international_number_and_servi
def test_validate_and_format_recipient_succeeds_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'
result = validate_and_format_recipient('+4407513332413', key_type, service_model, SMS_TYPE)
assert result == '+447513332413'
def test_validate_and_format_recipient_fails_when_no_recipient():