mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-16 20:48:37 -04:00
Merge pull request #1423 from alphagov/fix-firetext-inbound-500s
fix 500s when inbound msgs sent from alphanumerics rather than normal…
This commit is contained in:
@@ -1,28 +1,20 @@
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from flask import json
|
||||
from freezegun import freeze_time
|
||||
|
||||
from tests import create_authorization_header
|
||||
from tests.app.db import create_inbound_sms, create_service, create_service_with_inbound_number
|
||||
|
||||
|
||||
def test_get_inbound_sms_with_no_params(client, sample_service):
|
||||
def test_post_to_get_inbound_sms_with_no_params(admin_request, sample_service):
|
||||
one = create_inbound_sms(sample_service)
|
||||
two = create_inbound_sms(sample_service)
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
sms = json_resp['data']
|
||||
sms = admin_request.post(
|
||||
'inbound_sms.post_query_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data={}
|
||||
)['data']
|
||||
|
||||
assert len(sms) == 2
|
||||
assert {inbound['id'] for inbound in sms} == {str(one.id), str(two.id)}
|
||||
@@ -37,40 +29,34 @@ def test_get_inbound_sms_with_no_params(client, sample_service):
|
||||
}
|
||||
|
||||
|
||||
def test_get_inbound_sms_with_limit(client, sample_service):
|
||||
def test_post_to_get_inbound_sms_with_limit(admin_request, sample_service):
|
||||
with freeze_time('2017-01-01'):
|
||||
one = create_inbound_sms(sample_service)
|
||||
with freeze_time('2017-01-02'):
|
||||
two = create_inbound_sms(sample_service)
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {'limit': 1}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
sms = json_resp['data']
|
||||
sms = admin_request.post(
|
||||
'inbound_sms.post_query_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data=data
|
||||
)['data']
|
||||
|
||||
assert len(sms) == 1
|
||||
assert sms[0]['id'] == str(two.id)
|
||||
|
||||
|
||||
def test_get_inbound_sms_should_error_with_invalid_limit(client, sample_service):
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
def test_post_to_get_inbound_sms_should_error_with_invalid_limit(admin_request, sample_service):
|
||||
data = {'limit': 'limit'}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
error_resp = admin_request.post(
|
||||
'inbound_sms.post_query_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data=data,
|
||||
_expected_status=400
|
||||
)
|
||||
|
||||
error_resp = json.loads(response.get_data(as_text=True))
|
||||
assert error_resp['status_code'] == 400
|
||||
assert error_resp['errors'] == [{
|
||||
'error': 'ValidationError',
|
||||
@@ -78,73 +64,61 @@ def test_get_inbound_sms_should_error_with_invalid_limit(client, sample_service)
|
||||
}]
|
||||
|
||||
|
||||
def test_get_inbound_sms_should_error_with_invalid_phone_number(client, sample_service):
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {'phone_number': 'invalid phone number'}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
error_resp = json.loads(response.get_data(as_text=True))
|
||||
assert error_resp['status_code'] == 400
|
||||
assert error_resp['errors'] == [{
|
||||
'error': 'ValidationError',
|
||||
'message': "phone_number Must not contain letters or symbols"
|
||||
}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('user_number', [
|
||||
'(07700) 900-001',
|
||||
'+4407700900001',
|
||||
'447700900001',
|
||||
])
|
||||
def test_get_inbound_sms_filters_user_number(client, sample_service, user_number):
|
||||
def test_post_to_get_inbound_sms_filters_user_number(admin_request, sample_service, user_number):
|
||||
# user_number in the db is international and normalised
|
||||
one = create_inbound_sms(sample_service, user_number='447700900001')
|
||||
two = create_inbound_sms(sample_service, user_number='447700900002')
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {
|
||||
'limit': 1,
|
||||
'phone_number': user_number
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
sms = admin_request.post(
|
||||
'inbound_sms.post_query_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data=data
|
||||
)['data']
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
sms = json_resp['data']
|
||||
assert len(sms) == 1
|
||||
assert sms[0]['id'] == str(one.id)
|
||||
assert sms[0]['user_number'] == str(one.user_number)
|
||||
|
||||
|
||||
def test_get_inbound_sms_filters_international_user_number(admin_request, sample_service):
|
||||
def test_post_to_get_inbound_sms_filters_international_user_number(admin_request, sample_service):
|
||||
# user_number in the db is international and normalised
|
||||
one = create_inbound_sms(sample_service, user_number='12025550104')
|
||||
two = create_inbound_sms(sample_service)
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {
|
||||
'limit': 1,
|
||||
'phone_number': '+1 (202) 555-0104'
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
path='/service/{}/inbound-sms'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
sms = admin_request.post(
|
||||
'inbound_sms.post_query_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data=data
|
||||
)['data']
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
sms = json_resp['data']
|
||||
assert len(sms) == 1
|
||||
assert sms[0]['id'] == str(one.id)
|
||||
assert sms[0]['user_number'] == str(one.user_number)
|
||||
|
||||
|
||||
def test_post_to_get_inbound_sms_allows_badly_formatted_number(admin_request, sample_service):
|
||||
one = create_inbound_sms(sample_service, user_number='ALPHANUM3R1C')
|
||||
|
||||
sms = admin_request.post(
|
||||
'inbound_sms.get_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
_data={'phone_number': 'ALPHANUM3R1C'}
|
||||
)['data']
|
||||
|
||||
assert len(sms) == 1
|
||||
assert sms[0]['id'] == str(one.id)
|
||||
@@ -156,7 +130,7 @@ def test_get_inbound_sms_filters_international_user_number(admin_request, sample
|
||||
##############################################################
|
||||
|
||||
|
||||
def test_get_inbound_sms(admin_request, sample_service):
|
||||
def test_old_get_inbound_sms(admin_request, sample_service):
|
||||
one = create_inbound_sms(sample_service)
|
||||
two = create_inbound_sms(sample_service)
|
||||
|
||||
@@ -180,7 +154,7 @@ def test_get_inbound_sms(admin_request, sample_service):
|
||||
}
|
||||
|
||||
|
||||
def test_get_inbound_sms_limits(admin_request, sample_service):
|
||||
def test_old_get_inbound_sms_limits(admin_request, sample_service):
|
||||
with freeze_time('2017-01-01'):
|
||||
one = create_inbound_sms(sample_service)
|
||||
with freeze_time('2017-01-02'):
|
||||
@@ -201,7 +175,7 @@ def test_get_inbound_sms_limits(admin_request, sample_service):
|
||||
'+4407700900001',
|
||||
'447700900001',
|
||||
])
|
||||
def test_get_inbound_sms_filters_user_number(admin_request, sample_service, user_number):
|
||||
def test_old_get_inbound_sms_filters_user_number(admin_request, sample_service, user_number):
|
||||
# user_number in the db is international and normalised
|
||||
one = create_inbound_sms(sample_service, user_number='447700900001')
|
||||
two = create_inbound_sms(sample_service, user_number='447700900002')
|
||||
@@ -217,7 +191,7 @@ def test_get_inbound_sms_filters_user_number(admin_request, sample_service, user
|
||||
assert sms['data'][0]['user_number'] == str(one.user_number)
|
||||
|
||||
|
||||
def test_get_inbound_sms_filters_international_user_number(admin_request, sample_service):
|
||||
def test_old_get_inbound_sms_filters_international_user_number(admin_request, sample_service):
|
||||
# user_number in the db is international and normalised
|
||||
one = create_inbound_sms(sample_service, user_number='12025550104')
|
||||
two = create_inbound_sms(sample_service)
|
||||
@@ -233,6 +207,20 @@ def test_get_inbound_sms_filters_international_user_number(admin_request, sample
|
||||
assert sms['data'][0]['user_number'] == str(one.user_number)
|
||||
|
||||
|
||||
def test_old_get_inbound_sms_allows_badly_formatted_number(admin_request, sample_service):
|
||||
one = create_inbound_sms(sample_service, user_number='ALPHANUM3R1C')
|
||||
|
||||
sms = admin_request.get(
|
||||
'inbound_sms.get_inbound_sms_for_service',
|
||||
service_id=sample_service.id,
|
||||
user_number='ALPHANUM3R1C',
|
||||
)
|
||||
|
||||
assert len(sms['data']) == 1
|
||||
assert sms['data'][0]['id'] == str(one.id)
|
||||
assert sms['data'][0]['user_number'] == str(one.user_number)
|
||||
|
||||
|
||||
##############################
|
||||
# End delete section
|
||||
##############################
|
||||
|
||||
@@ -413,3 +413,24 @@ def test_firetext_inbound_sms_auth(notify_db_session, notify_api, client, mocker
|
||||
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
|
||||
|
||||
|
||||
def test_create_inbound_sms_object_works_with_alphanumeric_sender(sample_service_full_permissions):
|
||||
data = {
|
||||
'Message': 'hello',
|
||||
'Number': sample_service_full_permissions.get_inbound_number(),
|
||||
'MSISDN': 'ALPHANUM3R1C',
|
||||
'DateRecieved': '2017-01-02+03%3A04%3A05',
|
||||
'ID': 'bar',
|
||||
}
|
||||
|
||||
inbound_sms = create_inbound_sms_object(
|
||||
service=sample_service_full_permissions,
|
||||
content=format_mmg_message(data["Message"]),
|
||||
from_number='ALPHANUM3R1C',
|
||||
provider_ref='foo',
|
||||
date_received=None,
|
||||
provider_name="mmg"
|
||||
)
|
||||
|
||||
assert inbound_sms.user_number == 'ALPHANUM3R1C'
|
||||
|
||||
Reference in New Issue
Block a user