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

Merge branch 'master' into remove-initial-update-sms-sender

Browse Source
This commit is contained in:
Rebecca Law
2017-11-09 11:53:29 +00:00
parent 0054361044 4b99d6bb7a
commit 3713cc2353
19 changed files with 634 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
tests/app/authentication/test_authentication.py
View File

@@ -12,7 +12,7 @@ from notifications_python_client.authentication import create_jwt_token
from app import api_user
from app.dao.api_key_dao import get_unsigned_secrets, save_model_api_key, get_unsigned_secret, expire_api_key
from app.models import ApiKey, KEY_TYPE_NORMAL
from app.authentication.auth import restrict_ip_sms, AuthError
from app.authentication.auth import restrict_ip_sms, AuthError, check_route_secret
# Test the require_admin_auth and require_auth methods
@@ -372,3 +372,158 @@ def test_allow_valid_ips_bits(restrict_ip_sms_app):
)
assert response.status_code == 200
@pytest.fixture
def route_secret_app_with_key_1_only():
app = flask.Flask(__name__)
app.config['TESTING'] = True
app.config['ROUTE_SECRET_KEY_1'] = "key_1"
app.config['ROUTE_SECRET_KEY_2'] = ""
app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
blueprint = flask.Blueprint('route_secret_app_with_key_1_only', __name__)
@blueprint.route('/')
def test_endpoint():
return 'OK', 200
blueprint.before_request(check_route_secret)
app.register_blueprint(blueprint)
with app.test_request_context(), app.test_client() as client:
yield client
def test_route_secret_key_1_is_used(route_secret_app_with_key_1_only):
response = route_secret_app_with_key_1_only.get(
path='/',
headers=[
('X-Custom-forwarder', 'key_1'),
]
)
resp_json = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert resp_json['key_used'] == 1
# This tests for when we do key rotation when we use both keys
@pytest.fixture
def route_secret_app_both_keys():
app = flask.Flask(__name__)
app.config['TESTING'] = True
app.config['ROUTE_SECRET_KEY_1'] = "key_1"
app.config['ROUTE_SECRET_KEY_2'] = "key_2"
app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
blueprint = flask.Blueprint('route_secret_app_both_keys', __name__)
@blueprint.route('/')
def test_endpoint():
return 'OK', 200
blueprint.before_request(check_route_secret)
app.register_blueprint(blueprint)
with app.test_request_context(), app.test_client() as client:
yield client
@pytest.mark.parametrize('secret_header, expected_key_used', [
('key_2', 2),
('key_1', 1)
])
def test_can_use_either_secret_route_key(route_secret_app_both_keys, secret_header, expected_key_used):
print(secret_header)
response = route_secret_app_both_keys.get(
path='/',
headers=[
('X-Custom-forwarder', secret_header),
]
)
resp_json = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert resp_json['key_used'] == expected_key_used
@pytest.fixture
def route_secret_app_with_key_2_only():
app = flask.Flask(__name__)
app.config['TESTING'] = True
app.config['ROUTE_SECRET_KEY_1'] = ""
app.config['ROUTE_SECRET_KEY_2'] = "key_2"
app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
blueprint = flask.Blueprint('route_secret_app_with_key_2_only', __name__)
@blueprint.route('/')
def test_endpoint():
return 'OK', 200
blueprint.before_request(check_route_secret)
app.register_blueprint(blueprint)
with app.test_request_context(), app.test_client() as client:
yield client
def test_route_secret_key_2_is_used(route_secret_app_with_key_2_only):
response = route_secret_app_with_key_2_only.get(
path='/',
headers=[
('X-Custom-Forwarder', 'key_2'),
]
)
resp_json = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert resp_json['key_used'] == 2
# TODO: expected to fail because we have not implement blocking yet
@pytest.mark.parametrize('header_name, secret', [
pytest.mark.xfail(('some-header', 'some-value')),
pytest.mark.xfail(('X-Custom-Forwarder', 'wrong-value')),
])
def test_no_route_secret_raise_403(route_secret_app_with_key_2_only, header_name, secret):
response = route_secret_app_with_key_2_only.get(
path='/',
headers=[
(header_name, secret)
]
)
assert response.status_code == 403
@pytest.fixture
def route_secret_app_with_no_key():
app = flask.Flask(__name__)
app.config['TESTING'] = True
app.config['ROUTE_SECRET_KEY_1'] = ""
app.config['ROUTE_SECRET_KEY_2'] = ""
app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
blueprint = flask.Blueprint('route_secret_app_with_no_key', __name__)
@blueprint.route('/')
def test_endpoint():
return 'OK', 200
blueprint.before_request(check_route_secret)
app.register_blueprint(blueprint)
with app.test_request_context(), app.test_client() as client:
yield client
# TODO: expected to fail because we have not implement blocking yet
@pytest.mark.parametrize('secret', [
pytest.mark.xfail('some-header')
])
def test_route_secret_no_key_set_should_fail(route_secret_app_with_no_key, secret):
with pytest.raises(AuthError) as exc_info:
response = route_secret_app_with_no_key.get(
path='/',
headers=[
('X-Custom-Forwarder', 'some_value'),
]
)
resp_json = json.loads(response.get_data(as_text=True))
exc_info.value.short_message == 'X-Custom-Forwarder, no secret was set on server'
assert resp_json['key_used'] is None

20
tests/app/conftest.py
View File

@@ -864,6 +864,24 @@ def sms_code_template(notify_db,
)
@pytest.fixture(scope='function')
def email_2fa_code_template(notify_db, notify_db_session):
service, user = notify_service(notify_db, notify_db_session)
return create_custom_template(
service=service,
user=user,
template_config_name='EMAIL_2FA_TEMPLATE_ID',
content=(
'Hi ((name)),'
''
'To sign in to GOV.​UK Notify please open this link:'
'((url))'
),
subject='Sign in to GOV.UK Notify',
template_type='email'
)
@pytest.fixture(scope='function')
def email_verification_template(notify_db,
notify_db_session):
@@ -871,7 +889,7 @@ def email_verification_template(notify_db,
return create_custom_template(
service=service,
user=user,
template_config_name='EMAIL_VERIFY_CODE_TEMPLATE_ID',
template_config_name='NEW_USER_EMAIL_VERIFICATION_TEMPLATE_ID',
content='((user_name)) use ((url)) to complete registration',
template_type='email'
)

1
tests/app/dao/test_services_dao.py
View File

@@ -97,6 +97,7 @@ def test_create_service(sample_user):
assert service_db.branding == BRANDING_GOVUK
assert service_db.dvla_organisation_id == DVLA_ORG_HM_GOVERNMENT
assert service_db.research_mode is False
assert service_db.prefix_sms is True
assert service.active is True
assert sample_user in service_db.users

45
tests/app/delivery/test_send_to_providers.py
View File

@@ -671,23 +671,26 @@ def test_should_set_international_phone_number_to_sent_status(
assert notification.status == 'sent'
@pytest.mark.parametrize('sms_sender, expected_sender, expected_content', [
('foo', 'foo', 'bar'),
@pytest.mark.parametrize('sms_sender, expected_sender, prefix_sms, expected_content', [
('foo', 'foo', False, 'bar'),
('foo', 'foo', True, 'Sample service: bar'),
# if 40604 is actually in DB then treat that as if entered manually
('40604', '40604', 'bar'),
('40604', '40604', False, 'bar'),
# 'testing' is the FROM_NUMBER during unit tests
('testing', 'testing', 'Sample service: bar'),
('testing', 'testing', True, 'Sample service: bar'),
('testing', 'testing', False, 'bar'),
])
def test_should_handle_sms_sender_and_prefix_message(
mocker,
sms_sender,
prefix_sms,
expected_sender,
expected_content,
notify_db_session
):
mocker.patch('app.mmg_client.send_sms')
mocker.patch('app.delivery.send_to_providers.create_initial_notification_statistic_tasks')
service = create_service_with_defined_sms_sender(sms_sender_value=sms_sender)
service = create_service_with_defined_sms_sender(sms_sender_value=sms_sender, prefix_sms=prefix_sms)
template = create_template(service, content='bar')
notification = create_notification(template)
@@ -701,38 +704,6 @@ def test_should_handle_sms_sender_and_prefix_message(
)
@pytest.mark.parametrize('sms_sender, prefix_setting, expected_content', [
('foo', True, 'Sample service: bar'),
('foo', False, 'bar'),
('foo', None, 'bar'),
# 'testing' is the default SMS sender in unit tests
('testing', None, 'Sample service: bar'),
('testing', False, 'bar'),
])
def test_should_handle_sms_prefix_setting(
mocker,
sms_sender,
prefix_setting,
expected_content,
notify_db_session
):
mocker.patch('app.mmg_client.send_sms')
mocker.patch('app.delivery.send_to_providers.create_initial_notification_statistic_tasks')
service = create_service_with_defined_sms_sender(
sms_sender_value=sms_sender, prefix_sms=prefix_setting
)
template = create_template(service, content='bar')
notification = create_notification(template)
send_to_providers.send_sms_to_provider(notification)
mmg_client.send_sms.assert_called_once_with(
content=expected_content,
sender=ANY,
to=ANY,
reference=ANY,
)
def test_should_use_inbound_number_as_sender_if_default_sms_sender(
notify_db_session,
mocker

1
tests/app/invite/test_invite_rest.py
View File

@@ -115,6 +115,7 @@ def test_get_all_invited_users_by_service(client, notify_db, notify_db_session,
for invite in json_resp['data']:
assert invite['service'] == str(sample_service.id)
assert invite['from_user'] == str(invite_from.id)
assert invite['auth_type'] == SMS_AUTH_TYPE
assert invite['id']

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

@@ -8,7 +8,6 @@ import pytest
from flask import url_for, current_app
from freezegun import freeze_time
from app.dao.service_sms_sender_dao import dao_add_sms_sender_for_service
from app.dao.services_dao import dao_remove_user_from_service
from app.dao.templates_dao import dao_redact_template
from app.dao.users_dao import save_model_user
@@ -35,7 +34,9 @@ from tests.app.db import (
create_reply_to_email,
create_letter_contact,
create_inbound_number,
create_service_sms_sender
create_service_sms_sender,
create_service_with_defined_sms_sender,
create_service_with_inbound_number
)
from tests.app.db import create_user
from app.dao.date_util import get_current_financial_year_start_year
@@ -322,7 +323,7 @@ def test_create_service(client, sample_user):
service_sms_senders = ServiceSmsSender.query.filter_by(service_id=service_db.id).all()
assert len(service_sms_senders) == 1
assert service_sms_senders[0].sms_sender == service_db.get_default_sms_sender()
assert service_sms_senders[0].sms_sender == current_app.config['FROM_NUMBER']
def test_should_not_create_service_with_missing_user_id_field(notify_api, fake_uuid):
@@ -1505,24 +1506,17 @@ def test_get_only_api_created_notifications_for_service(
assert resp['notifications'][0]['id'] == str(without_job.id)
@pytest.mark.parametrize('default_sms_sender, should_prefix', [
(None, True), # None means use default
('Foo', False),
@pytest.mark.parametrize('should_prefix', [
True,
False,
])
def test_prefixing_messages_based_on_sms_sender(
def test_prefixing_messages_based_on_prefix_sms(
client,
notify_db_session,
default_sms_sender,
should_prefix,
):
service = create_service()
if default_sms_sender:
# add another sms sender that is the default.
dao_add_sms_sender_for_service(service_id=service.id, sms_sender=default_sms_sender, is_default=True)
create_service_sms_sender(
service=service,
sms_sender='ignored',
is_default=False,
service = create_service(
prefix_sms=should_prefix
)
result = client.get(
@@ -2632,7 +2626,7 @@ def test_add_service_sms_sender_can_add_multiple_senders(client, notify_db_sessi
def test_add_service_sms_sender_when_it_is_an_inbound_number_updates_the_only_existing_sms_sender(
client, notify_db_session):
service = create_service()
service = create_service_with_defined_sms_sender(sms_sender_value='GOVUK')
inbound_number = create_inbound_number(number='12345')
data = {
"sms_sender": str(inbound_number.id),
@@ -2657,7 +2651,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_updates_the_only_ex
def test_add_service_sms_sender_when_it_is_an_inbound_number_inserts_new_sms_sender_when_more_than_one(
client, notify_db_session):
service = create_service()
service = create_service_with_defined_sms_sender(sms_sender_value='GOVUK')
create_service_sms_sender(service=service, sms_sender="second", is_default=False)
inbound_number = create_inbound_number(number='12345')
data = {
@@ -2682,7 +2676,7 @@ def test_add_service_sms_sender_when_it_is_an_inbound_number_inserts_new_sms_sen
def test_add_service_sms_sender_switches_default(client, notify_db_session):
service = create_service()
service = create_service_with_defined_sms_sender(sms_sender_value='first')
data = {
"sms_sender": 'second',
"is_default": True,
@@ -2696,7 +2690,7 @@ def test_add_service_sms_sender_switches_default(client, notify_db_session):
assert resp_json['sms_sender'] == 'second'
assert not resp_json['inbound_number_id']
assert resp_json['is_default']
sms_senders = ServiceSmsSender.query.filter_by(sms_sender='testing').first()
sms_senders = ServiceSmsSender.query.filter_by(sms_sender='first').first()
assert not sms_senders.is_default
@@ -2734,7 +2728,7 @@ def test_update_service_sms_sender(client, notify_db_session):
def test_update_service_sms_sender_switches_default(client, notify_db_session):
service = create_service()
service = create_service_with_defined_sms_sender(sms_sender_value='first')
service_sms_sender = create_service_sms_sender(service=service, sms_sender='1235', is_default=False)
data = {
"sms_sender": 'second',
@@ -2749,7 +2743,7 @@ def test_update_service_sms_sender_switches_default(client, notify_db_session):
assert resp_json['sms_sender'] == 'second'
assert not resp_json['inbound_number_id']
assert resp_json['is_default']
sms_senders = ServiceSmsSender.query.filter_by(sms_sender='testing').first()
sms_senders = ServiceSmsSender.query.filter_by(sms_sender='first').first()
assert not sms_senders.is_default
@@ -2809,10 +2803,8 @@ def test_get_service_sms_sender_by_id_returns_404_when_service_does_not_exist(cl
def test_get_service_sms_sender_by_id_returns_404_when_sms_sender_does_not_exist(client, notify_db_session):
service_sms_sender = create_service_sms_sender(service=create_service(),
sms_sender='1235',
is_default=False)
response = client.get('/service/{}/sms-sender/{}'.format(service_sms_sender.service_id, uuid.uuid4()),
service = create_service()
response = client.get('/service/{}/sms-sender/{}'.format(service.id, uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert response.status_code == 404

4
tests/app/test_cloudfoundry_config.py
View File

@@ -16,7 +16,9 @@ def notify_config():
'admin_client_secret': 'admin client secret',
'secret_key': 'secret key',
'dangerous_salt': 'dangerous salt',
'allow_ip_inbound_sms': ['111.111.111.111', '100.100.100.100']
'allow_ip_inbound_sms': ['111.111.111.111', '100.100.100.100'],
'route_secret_key_1': "key_1",
'route_secret_key_2': ""
}
}

152
tests/app/user/test_rest_verify.py
View File

@@ -9,11 +9,14 @@ import pytest
from flask import url_for, current_app
from freezegun import freeze_time
from app.dao.users_dao import create_user_code
from app.dao.services_dao import dao_update_service, dao_fetch_service_by_id
from app.models import (
VerifyCode,
Notification,
User,
Notification
VerifyCode,
EMAIL_TYPE,
SMS_TYPE
)
from app import db
import app.celery.tasks
@@ -40,25 +43,6 @@ def test_user_verify_sms_code(client, sample_sms_code):
assert sample_sms_code.user.current_session_id is not None
@freeze_time('2016-01-01T12:00:00')
def test_user_verify_email_code(client, sample_email_code):
sample_email_code.user.logged_in_at = datetime.utcnow() - timedelta(days=1)
assert not VerifyCode.query.first().code_used
assert sample_email_code.user.current_session_id is None
data = json.dumps({
'code_type': sample_email_code.code_type,
'code': sample_email_code.txt_code})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.verify_user_code', user_id=sample_email_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert VerifyCode.query.first().code_used
assert sample_email_code.user.logged_in_at == datetime.utcnow() - timedelta(days=1)
assert sample_email_code.user.current_session_id is None
def test_user_verify_code_missing_code(client,
sample_sms_code):
assert not VerifyCode.query.first().code_used
@@ -201,17 +185,15 @@ def test_send_user_sms_code(client,
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
url_for('user.send_user_2fa_code', code_type='sms', user_id=sample_user.id),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert mocked.call_count == 1
assert VerifyCode.query.count() == 1
assert VerifyCode.query.first().check_code('11111')
assert VerifyCode.query.one().check_code('11111')
assert Notification.query.count() == 1
notification = Notification.query.first()
notification = Notification.query.one()
assert notification.personalisation == {'verify_code': '11111'}
assert notification.to == sample_user.mobile_number
assert str(notification.service_id) == current_app.config['NOTIFY_SERVICE_ID']
@@ -236,7 +218,7 @@ def test_send_user_code_for_sms_with_optional_to_field(client,
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
url_for('user.send_user_2fa_code', code_type='sms', user_id=sample_user.id),
data=json.dumps({'to': to_number}),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -254,7 +236,7 @@ def test_send_sms_code_returns_404_for_bad_input_data(client):
uuid_ = uuid.uuid4()
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_sms_code', user_id=uuid_),
url_for('user.send_user_2fa_code', code_type='sms', user_id=uuid_),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
@@ -275,25 +257,26 @@ def test_send_sms_code_returns_204_when_too_many_codes_already_created(client, s
assert VerifyCode.query.count() == 10
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
url_for('user.send_user_2fa_code', code_type='sms', user_id=sample_user.id),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert VerifyCode.query.count() == 10
def test_send_user_email_verification(client,
sample_user,
mocker,
email_verification_template):
def test_send_new_user_email_verification(client,
sample_user,
mocker,
email_verification_template):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_email_verification', user_id=str(sample_user.id)),
url_for('user.send_new_user_email_verification', user_id=str(sample_user.id)),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
notification = Notification.query.first()
assert VerifyCode.query.count() == 0
mocked.assert_called_once_with(([str(notification.id)]), queue="notify-internal-tasks")
@@ -305,7 +288,7 @@ def test_send_email_verification_returns_404_for_bad_input_data(client, notify_d
uuid_ = uuid.uuid4()
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_email_verification', user_id=uuid_),
url_for('user.send_new_user_email_verification', user_id=uuid_),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
@@ -355,3 +338,102 @@ def test_reset_failed_login_count_returns_404_when_user_does_not_exist(client):
data={},
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert resp.status_code == 404
def test_send_user_email_code(admin_request, mocker, sample_user, email_2fa_code_template):
deliver_email = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
data = {
'to': None
}
admin_request.post(
'user.send_user_2fa_code',
code_type='email',
user_id=sample_user.id,
_data=data,
_expected_status=204
)
noti = Notification.query.one()
assert noti.to == sample_user.email_address
assert str(noti.template_id) == current_app.config['EMAIL_2FA_TEMPLATE_ID']
assert noti.personalisation['name'] == 'Test User'
deliver_email.assert_called_once_with(
[str(noti.id)],
queue='notify-internal-tasks'
)
def test_send_user_email_code_with_urlencoded_next_param(admin_request, mocker, sample_user, email_2fa_code_template):
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
data = {
'to': None,
'next': '/services'
}
admin_request.post(
'user.send_user_2fa_code',
code_type='email',
user_id=sample_user.id,
_data=data,
_expected_status=204
)
noti = Notification.query.one()
code = VerifyCode.query.one()
assert noti.personalisation['url'].endswith('?next=%2Fservices')
def test_send_email_code_returns_404_for_bad_input_data(admin_request):
resp = admin_request.post(
'user.send_user_2fa_code',
code_type='email',
user_id=uuid.uuid4(),
_data={},
_expected_status=404
)
assert resp['message'] == 'No result found'
@freeze_time('2016-01-01T12:00:00')
def test_user_verify_email_code(admin_request, sample_user):
magic_code = str(uuid.uuid4())
verify_code = create_user_code(sample_user, magic_code, EMAIL_TYPE)
data = {
'code_type': 'email',
'code': magic_code
}
admin_request.post(
'user.verify_user_code',
user_id=sample_user.id,
_data=data,
_expected_status=204
)
assert verify_code.code_used
assert sample_user.logged_in_at == datetime.utcnow()
assert sample_user.current_session_id is not None
@pytest.mark.parametrize('code_type', [EMAIL_TYPE, SMS_TYPE])
@freeze_time('2016-01-01T12:00:00')
def test_user_verify_email_code_fails_if_code_already_used(admin_request, sample_user, code_type):
magic_code = str(uuid.uuid4())
verify_code = create_user_code(sample_user, magic_code, code_type)
verify_code.code_used = True
data = {
'code_type': code_type,
'code': magic_code
}
admin_request.post(
'user.verify_user_code',
user_id=sample_user.id,
_data=data,
_expected_status=400
)
assert verify_code.code_used
assert sample_user.logged_in_at is None
assert sample_user.current_session_id is None