darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-09 23:02:13 -05:00
Code Issues Packages Projects Releases Wiki Activity

get_inbound_sms queries from the admin should also allow alphanumerics.

Browse Source 
Refactored the call to make the POST and the GET versions of the method
much more distinct.
This commit is contained in:
Leo Hemsted
2017-11-23 16:46:39 +00:00
parent f3d129b0d8
commit 3daf039fbe
3 changed files with 89 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/inbound_sms/inbound_sms_schemas.py
View File

@@ -3,7 +3,7 @@ get_inbound_sms_for_service_schema = {
"description": "schema for parameters allowed when searching for to field=",
"type": "object",
"properties": {
"phone_number": {"type": "string", "format": "phone_number"},
"phone_number": {"type": "string"},
"limit": {"type": ["integer", "null"], "minimum": 1}
}
}

47
app/inbound_sms/rest.py
View File

@@ -1,11 +1,10 @@
from flask import (
Blueprint,
jsonify,
request,
current_app, json)
from jsonschema import ValidationError
request
)
from notifications_utils.recipients import validate_and_format_phone_number
from notifications_utils.recipients import try_validate_and_format_phone_number
from app.dao.inbound_sms_dao import (
dao_get_inbound_sms_for_service,
@@ -26,25 +25,31 @@ inbound_sms = Blueprint(
register_errors(inbound_sms)
@inbound_sms.route('', methods=['POST', 'GET'])
def get_inbound_sms_for_service(service_id):
if request.method == 'GET':
limit = request.args.get('limit')
user_number = request.args.get('user_number')
if user_number:
# we use this to normalise to an international phone number
user_number = validate_and_format_phone_number(user_number, international=True)
results = dao_get_inbound_sms_for_service(service_id, limit, user_number)
return jsonify(data=[row.serialize() for row in results])
@inbound_sms.route('', methods=['POST'])
def post_query_inbound_sms_for_service(service_id):
form = validate(request.get_json(), get_inbound_sms_for_service_schema)
if 'phone_number' in form:
# we use this to normalise to an international phone number - but this may fail if it's an alphanumeric
user_number = try_validate_and_format_phone_number(form['phone_number'], international=True)
else:
form = validate(request.get_json(), get_inbound_sms_for_service_schema)
results = dao_get_inbound_sms_for_service(service_id, form.get('limit'), form.get('phone_number'))
user_number = None
results = dao_get_inbound_sms_for_service(service_id, form.get('limit'), user_number)
return jsonify(data=[row.serialize() for row in results])
return jsonify(data=[row.serialize() for row in results])
@inbound_sms.route('', methods=['GET'])
def get_inbound_sms_for_service(service_id):
limit = request.args.get('limit')
user_number = request.args.get('user_number')
if user_number:
# we use this to normalise to an international phone number - but this may fail if it's an alphanumeric
user_number = try_validate_and_format_phone_number(user_number, international=True)
results = dao_get_inbound_sms_for_service(service_id, limit, user_number)
return jsonify(data=[row.serialize() for row in results])
@inbound_sms.route('/summary')

136
tests/app/inbound_sms/test_rest.py
View File

@@ -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
##############################