mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-20 23:41:17 -05:00
e569c54f45
This means that these codes won't be delayed by large jobs going through the send-sms/email queues. send_user_sms_code now works much more like the endpoints for sending notifications, by persisting the notification and only using the deliver_sms task (instead of using send_sms as well). The workers consuming the `notify` queue should be able to handle the deliver task as well, so no change should be needed to the celery workers to support this. I think there's also a change in behaviour here: previously, if the Notify service was in research mode, 2FA codes would not have been sent out, making it impossible to log into the admin. Now, a call to this endpoint will always send out the notification even if we've put the Notify service into research mode, since we set the notification's key type to normal and ignore the service's research mode setting when sending the notification to the queue.
366 lines
15 KiB
Python
366 lines
15 KiB
Python
import json
|
|
import moto
|
|
import pytest
|
|
|
|
from datetime import (
|
|
datetime,
|
|
timedelta
|
|
)
|
|
|
|
from flask import url_for, current_app
|
|
from app.dao.services_dao import dao_update_service, dao_fetch_service_by_id
|
|
from app.models import (
|
|
VerifyCode,
|
|
User,
|
|
Notification
|
|
)
|
|
|
|
from app import db, encryption
|
|
|
|
from tests import create_authorization_header
|
|
from freezegun import freeze_time
|
|
|
|
import app.celery.tasks
|
|
|
|
|
|
def test_user_verify_code_sms(notify_api,
|
|
sample_sms_code):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/code'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
assert not VerifyCode.query.first().code_used
|
|
data = json.dumps({
|
|
'code_type': sample_sms_code.code_type,
|
|
'code': sample_sms_code.txt_code})
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 204
|
|
assert VerifyCode.query.first().code_used
|
|
|
|
|
|
def test_user_verify_code_sms_missing_code(notify_api,
|
|
sample_sms_code):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/code'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
assert not VerifyCode.query.first().code_used
|
|
data = json.dumps({'code_type': sample_sms_code.code_type})
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 400
|
|
assert not VerifyCode.query.first().code_used
|
|
|
|
|
|
@moto.mock_sqs
|
|
def test_user_verify_code_email(notify_api,
|
|
sqs_client_conn,
|
|
sample_email_code):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/code'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
assert not VerifyCode.query.first().code_used
|
|
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
|
|
|
|
|
|
def test_user_verify_code_email_bad_code(notify_api,
|
|
sample_email_code):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/code'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
assert not VerifyCode.query.first().code_used
|
|
data = json.dumps({
|
|
'code_type': sample_email_code.code_type,
|
|
'code': "blah"})
|
|
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 == 404
|
|
assert not VerifyCode.query.first().code_used
|
|
|
|
|
|
def test_user_verify_code_email_expired_code(notify_api,
|
|
sample_email_code):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/code'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
assert not VerifyCode.query.first().code_used
|
|
sample_email_code.expiry_datetime = (
|
|
datetime.utcnow() - timedelta(hours=1))
|
|
db.session.add(sample_email_code)
|
|
db.session.commit()
|
|
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 == 400
|
|
assert not VerifyCode.query.first().code_used
|
|
|
|
|
|
@freeze_time("2016-01-01 10:00:00.000000")
|
|
def test_user_verify_password(notify_api,
|
|
notify_db,
|
|
notify_db_session,
|
|
sample_user):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/password'
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({'password': 'password'})
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.verify_user_password', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 204
|
|
User.query.get(sample_user.id).logged_in_at == datetime.utcnow()
|
|
|
|
|
|
def test_user_verify_password_invalid_password(notify_api,
|
|
sample_user):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/password' invalid endpoint.
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({'password': 'bad password'})
|
|
auth_header = create_authorization_header()
|
|
|
|
assert sample_user.failed_login_count == 0
|
|
|
|
resp = client.post(
|
|
url_for('user.verify_user_password', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 400
|
|
json_resp = json.loads(resp.get_data(as_text=True))
|
|
assert 'Incorrect password' in json_resp['message']['password']
|
|
assert sample_user.failed_login_count == 1
|
|
|
|
|
|
def test_user_verify_password_valid_password_resets_failed_logins(notify_api,
|
|
sample_user):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({'password': 'bad password'})
|
|
auth_header = create_authorization_header()
|
|
|
|
assert sample_user.failed_login_count == 0
|
|
|
|
resp = client.post(
|
|
url_for('user.verify_user_password', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 400
|
|
json_resp = json.loads(resp.get_data(as_text=True))
|
|
assert 'Incorrect password' in json_resp['message']['password']
|
|
|
|
assert sample_user.failed_login_count == 1
|
|
|
|
data = json.dumps({'password': 'password'})
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.verify_user_password', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert resp.status_code == 204
|
|
assert sample_user.failed_login_count == 0
|
|
|
|
|
|
def test_user_verify_password_missing_password(notify_api,
|
|
sample_user):
|
|
"""
|
|
Tests POST endpoint '/<user_id>/verify/password' missing password.
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({'bingo': 'bongo'})
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.verify_user_password', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 400
|
|
json_resp = json.loads(resp.get_data(as_text=True))
|
|
assert 'Required field missing data' in json_resp['message']['password']
|
|
|
|
|
|
@pytest.mark.parametrize('research_mode', [True, False])
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_send_user_sms_code(notify_api,
|
|
sample_user,
|
|
sms_code_template,
|
|
mocker,
|
|
research_mode):
|
|
"""
|
|
Tests POST endpoint /user/<user_id>/sms-code
|
|
"""
|
|
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
if research_mode:
|
|
notify_service = dao_fetch_service_by_id(current_app.config['NOTIFY_SERVICE_ID'])
|
|
notify_service.research_mode = True
|
|
dao_update_service(notify_service)
|
|
|
|
data = json.dumps({})
|
|
auth_header = create_authorization_header()
|
|
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
|
|
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),
|
|
data=data,
|
|
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 Notification.query.count() == 1
|
|
notification = Notification.query.first()
|
|
assert notification.personalisation == {'verify_code': '11111'}
|
|
assert notification.to == sample_user.mobile_number
|
|
assert str(notification.service_id) == current_app.config['NOTIFY_SERVICE_ID']
|
|
|
|
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
|
|
([str(notification.id)]),
|
|
queue="notify"
|
|
)
|
|
|
|
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_send_user_code_for_sms_with_optional_to_field(notify_api,
|
|
sample_user,
|
|
sms_code_template,
|
|
mocker):
|
|
"""
|
|
Tests POST endpoint /user/<user_id>/sms-code with optional to field
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
to_number = '+441119876757'
|
|
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
|
|
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
|
|
data = json.dumps({'to': to_number})
|
|
auth_header = create_authorization_header()
|
|
|
|
resp = client.post(
|
|
url_for('user.send_user_sms_code', user_id=sample_user.id),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert resp.status_code == 204
|
|
assert mocked.call_count == 1
|
|
notification = Notification.query.first()
|
|
assert notification.to == to_number
|
|
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
|
|
([str(notification.id)]),
|
|
queue="notify"
|
|
)
|
|
|
|
|
|
def test_send_sms_code_returns_404_for_bad_input_data(notify_api, notify_db, notify_db_session):
|
|
"""
|
|
Tests POST endpoint /user/<user_id>/sms-code return 404 for bad input data
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({})
|
|
import uuid
|
|
uuid_ = uuid.uuid4()
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.send_user_sms_code', user_id=uuid_),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 404
|
|
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
|
|
|
|
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_send_user_email_verification(notify_api,
|
|
sample_user,
|
|
mocker,
|
|
email_verification_template):
|
|
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({})
|
|
mocker.patch('uuid.uuid4', return_value='some_uuid') # for the notification id
|
|
mocked = mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
mocker.patch('notifications_utils.url_safe_token.generate_token', return_value='the-token')
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.send_user_email_verification', user_id=str(sample_user.id)),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 204
|
|
assert mocked.call_count == 1
|
|
message = {
|
|
'template': str(email_verification_template.id),
|
|
'template_version': email_verification_template.version,
|
|
'to': sample_user.email_address,
|
|
'personalisation': {
|
|
'name': sample_user.name,
|
|
'url': current_app.config['ADMIN_BASE_URL'] + '/verify-email/' + 'the-token'
|
|
}
|
|
}
|
|
app.celery.tasks.send_email.apply_async.assert_called_once_with(
|
|
(str(current_app.config['NOTIFY_SERVICE_ID']),
|
|
'some_uuid',
|
|
encryption.encrypt(message),
|
|
"2016-01-01T11:09:00.061258Z"),
|
|
queue="notify")
|
|
|
|
|
|
def test_send_email_verification_returns_404_for_bad_input_data(notify_api, notify_db, notify_db_session):
|
|
"""
|
|
Tests POST endpoint /user/<user_id>/sms-code return 404 for bad input data
|
|
"""
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
data = json.dumps({})
|
|
import uuid
|
|
uuid_ = uuid.uuid4()
|
|
auth_header = create_authorization_header()
|
|
resp = client.post(
|
|
url_for('user.send_user_email_verification', user_id=uuid_),
|
|
data=data,
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert resp.status_code == 404
|
|
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
|