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

Refactor user/<user_id>/code into two endpoints.

Browse Source 
- Created new endpoint user/<user_id>/sms-code to send the sms verification code to the user.
- Create new endpoirtn user/<user_id>/email-code to send the email verifcation code to the user.
- Marked the old methods, schema, tests with a TODO to be deleted when the admin app is no longer sending messages to /user/<user_id>/code
- Added error handlers for DataError and NoResultFound. Data error catches invalid input errors.
- Added error handler for SqlAlchemyError which catches any other database errors.
- Removed the need for the try catches around the db calls in the user endpoints with the addition of the db error handlers.
- We may want to wrap db excpetions in the dao, if we want the No results found message to be more specific and say no result found for user.
This commit is contained in:
Rebecca Law
2016-02-19 11:37:35 +00:00
parent 9ea2acfdae
commit 17d14f291e
5 changed files with 228 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/app/user/test_rest.py
View File

@@ -261,7 +261,7 @@ def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_us
assert User.query.count() == 2
user = User.query.filter_by(id=sample_user.id).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp == {'result': 'error', 'message': 'User not found'}
assert json_resp == {'error': 'No result found'}
assert user == sample_user
assert user.email_address != new_email
@@ -318,8 +318,7 @@ def test_get_user_service(notify_api, notify_db, notify_db_session, sample_servi
assert json_resp['data']['id'] == str(another_service.id)
def test_get_user_service_user_not_exists(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
def test_get_user_service_user_not_exists(notify_api, sample_service, sample_admin_service_id):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" 404 is returned for invalid user.
"""
@@ -330,28 +329,25 @@ def test_get_user_service_user_not_exists(notify_api, notify_db, notify_db_sessi
path=url_for('user.get_service_by_user_id', user_id="123423",
service_id=sample_service.id),
method='GET')
print('** service users{}'.format(sample_service.users[0].id))
resp = client.get(
url_for('user.get_service_by_user_id', user_id="123423", service_id=sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "User not found" in json_resp['message']
assert json_resp == {'error': 'No result found'}
def test_get_user_service_service_not_exists(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
def test_get_user_service_service_not_exists(notify_api, sample_service):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" 404 is returned for invalid service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
assert Service.query.count() == 2
assert Service.query.count() == 1
import uuid
missing_service_id = uuid.uuid4()
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_service_by_user_id', user_id=user.id,
auth_header = create_authorization_header(path=url_for('user.get_service_by_user_id', user_id=user.id,
service_id=missing_service_id),
method='GET')
resp = client.get(
@@ -359,10 +355,10 @@ def test_get_user_service_service_not_exists(notify_api, notify_db, notify_db_se
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "Service not found" in json_resp['message']
assert json_resp == {'error': 'No result found'}
def test_delete_user(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
def test_delete_user(notify_api, sample_user, sample_admin_service_id):
"""
Tests DELETE endpoint '/<user_id>' delete user.
"""
@@ -407,3 +403,5 @@ def test_delete_user_not_exists(notify_api, notify_db, notify_db_session, sample
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert User.query.count() == 2
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp == {'error': 'No result found'}

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

@@ -2,9 +2,7 @@ import json
import moto
from datetime import (datetime, timedelta)
from flask import url_for
from app.models import (VerifyCode)
import app.celery.tasks
from app import db, encryption
from tests import create_authorization_header
@@ -191,7 +189,6 @@ def test_user_verify_password_valid_password_resets_failed_logins(notify_api,
notify_db,
notify_db_session,
sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({'password': 'bad password'})
@@ -249,9 +246,10 @@ def test_user_verify_password_missing_password(notify_api,
assert 'Required field missing data' in json_resp['message']['password']
# TODO: Remove this test once the admin app has stopped using it.
def test_send_user_code_for_sms(notify_api,
sample_sms_code,
mock_secret_code,
mock_encryption,
mock_celery_send_sms_code):
"""
Tests POST endpoint '/<user_id>/code' successful sms
@@ -269,35 +267,11 @@ def test_send_user_code_for_sms(notify_api,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
encrpyted = encryption.encrypt({'to': sample_sms_code.user.mobile_number, 'secret_code': '11111'})
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with([encrpyted], queue='sms-code')
def test_send_user_code_for_sms_with_optional_to_field(notify_api,
sample_sms_code,
mock_secret_code,
mock_celery_send_sms_code):
"""
Tests POST endpoint '/<user_id>/code' successful sms with optional to field
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({'code_type': 'sms', 'to': '+441119876757'})
auth_header = create_authorization_header(
path=url_for('user.send_user_code', user_id=sample_sms_code.user.id),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_code', user_id=sample_sms_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
encrypted = encryption.encrypt({'to': '+441119876757', 'secret_code': '11111'})
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with([encrypted], queue='sms-code')
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with(['something_encrypted'],
queue='sms-code')
# TODO: Remove this test once the admin app has stopped using it.
def test_send_user_code_for_email(notify_api,
sample_email_code,
mock_secret_code,
@@ -323,6 +297,7 @@ def test_send_user_code_for_email(notify_api,
queue='email-code')
# TODO: Remove this test once the admin app has stopped using it.
def test_send_user_code_for_email_uses_optional_to_field(notify_api,
sample_email_code,
mock_secret_code,
@@ -348,16 +323,130 @@ def test_send_user_code_for_email_uses_optional_to_field(notify_api,
queue='email-code')
def test_request_verify_code_schema_invalid_code_type(notify_api, notify_db, notify_db_session, sample_user):
from app.schemas import request_verify_code_schema
# TODO: Remove this test once the admin app has stopped using it.
def test_request_verify_code_schema_invalid_code_type(notify_api, sample_user):
from app.schemas import old_request_verify_code_schema
data = json.dumps({'code_type': 'not_sms'})
code, error = request_verify_code_schema.loads(data)
code, error = old_request_verify_code_schema.loads(data)
assert error == {'code_type': ['Invalid code type']}
def test_request_verify_code_schema_with_to(notify_api, notify_db, notify_db_session, sample_user):
from app.schemas import request_verify_code_schema
# TODO: Remove this method once the admin app has stopped using it.
def test_request_verify_code_schema_with_to(notify_api, sample_user):
from app.schemas import old_request_verify_code_schema
data = json.dumps({'code_type': 'sms', 'to': 'some@one.gov.uk'})
code, error = request_verify_code_schema.loads(data)
code, error = old_request_verify_code_schema.loads(data)
assert code == {'code_type': 'sms', 'to': 'some@one.gov.uk'}
assert error == {}
def test_send_user_sms_code(notify_api,
sample_sms_code,
mock_celery_send_sms_code,
mock_encryption):
"""
Tests POST endpoint /user/<user_id>/sms-code
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({})
auth_header = create_authorization_header(
path=url_for('user.send_user_sms_code', user_id=sample_sms_code.user.id),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_sms_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
print(resp.get_data(as_text=True))
assert resp.status_code == 204
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with(['something_encrypted'],
queue='sms-code')
def test_send_user_code_for_sms_with_optional_to_field(notify_api,
sample_sms_code,
mock_secret_code,
mock_celery_send_sms_code):
"""
Tests POST endpoint '/<user_id>/code' successful sms with optional to field
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({'code_type': 'sms', 'to': '+441119876757'})
auth_header = create_authorization_header(
path=url_for('user.send_user_sms_code', user_id=sample_sms_code.user.id),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_sms_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
encrypted = encryption.encrypt({'to': '+441119876757', 'secret_code': '11111'})
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with([encrypted], queue='sms-code')
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(
path=url_for('user.send_user_sms_code', user_id=uuid_),
method='POST',
request_body=data)
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))['error'] == 'No result found'
def test_send_user_email_code(notify_api,
sample_email_code,
mock_celery_send_email_code,
mock_encryption):
"""
Tests POST endpoint /user/<user_id>/email-code
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({})
auth_header = create_authorization_header(
path=url_for('user.send_user_email_code', user_id=sample_email_code.user.id),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_email_code', user_id=sample_email_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
print(resp.get_data(as_text=True))
assert resp.status_code == 204
app.celery.tasks.send_email_code.apply_async.assert_called_once_with(['something_encrypted'],
queue='email-code')
def test_send_user_email_code_returns_404_for_when_user_does_not_exist(notify_api, notify_db, notify_db_session):
"""
Tests POST endpoint /user/<user_id>/email-code return 404 for missing user
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({})
auth_header = create_authorization_header(
path=url_for('user.send_user_email_code', user_id=1),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_email_code', user_id=1),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert json.loads(resp.get_data(as_text=True))['error'] == 'No result found'