darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-09-09 17:45:41 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #834 from alphagov/reset-2fa

Browse Source 
ensure we reset failed_login_count when appropriate
This commit is contained in:
Leo Hemsted
2017-02-17 11:45:10 +00:00
committed by GitHub
parent ce1fac63ec 0515c51475
commit 1dbec34cf9
4 changed files with 376 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/dao/users_dao.py
View File

@@ -113,6 +113,7 @@ def reset_failed_login_count(user):
def update_user_password(user, password): def update_user_password(user, password):
# reset failed login count - they've just reset their password so should be fine
user.password = password user.password = password
user.password_changed_at = datetime.utcnow() user.password_changed_at = datetime.utcnow()
db.session.add(user) db.session.add(user)

14
app/user/rest.py
View File

@@ -59,12 +59,14 @@ def update_user(user_id):
user_to_update = get_user_by_id(user_id=user_id) user_to_update = get_user_by_id(user_id=user_id)
req_json = request.get_json() req_json = request.get_json()
update_dct, errors = user_schema_load_json.load(req_json) update_dct, errors = user_schema_load_json.load(req_json)
# TODO don't let password be updated in this PUT method (currently used by the forgot password flow)
pwd = req_json.get('password', None) pwd = req_json.get('password', None)
# TODO password validation, it is already done on the admin app if pwd is not None:
# but would be good to have the same validation here. if not pwd:
if pwd is not None and not pwd: errors.update({'password': ['Invalid data for field']})
errors.update({'password': ['Invalid data for field']}) raise InvalidRequest(errors, status_code=400)
raise InvalidRequest(errors, status_code=400) else:
reset_failed_login_count(user_to_update)
save_model_user(user_to_update, update_dict=update_dct, pwd=pwd) save_model_user(user_to_update, update_dict=update_dct, pwd=pwd)
return jsonify(data=user_schema.dump(user_to_update).data), 200 return jsonify(data=user_schema.dump(user_to_update).data), 200
@@ -130,6 +132,7 @@ def verify_user_code(user_id):
increment_failed_login_count(user_to_verify) increment_failed_login_count(user_to_verify)
raise InvalidRequest("Code has expired", status_code=400) raise InvalidRequest("Code has expired", status_code=400)
use_user_code(code.id) use_user_code(code.id)
reset_failed_login_count(user_to_verify)
return jsonify({}), 204 return jsonify({}), 204
@@ -323,6 +326,7 @@ def update_password(user_id):
if errors: if errors:
raise InvalidRequest(errors, status_code=400) raise InvalidRequest(errors, status_code=400)
reset_failed_login_count(user)
update_user_password(user, pwd) update_user_password(user, pwd)
return jsonify(data=user_schema.dump(user).data), 200 return jsonify(data=user_schema.dump(user).data), 200

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

@@ -10,174 +10,165 @@ from app.dao.permissions_dao import default_service_permissions
from tests import create_authorization_header from tests import create_authorization_header
def test_get_user_list(notify_api, notify_db, notify_db_session, sample_service): def test_get_user_list(client, sample_service):
""" """
Tests GET endpoint '/' to retrieve entire user list. Tests GET endpoint '/' to retrieve entire user list.
""" """
with notify_api.test_request_context(): header = create_authorization_header()
with notify_api.test_client() as client: response = client.get(url_for('user.get_user'),
header = create_authorization_header() headers=[header])
response = client.get(url_for('user.get_user'), assert response.status_code == 200
headers=[header]) json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200 assert len(json_resp['data']) == 1
json_resp = json.loads(response.get_data(as_text=True)) sample_user = sample_service.users[0]
assert len(json_resp['data']) == 1 expected_permissions = default_service_permissions
sample_user = sample_service.users[0] fetched = json_resp['data'][0]
expected_permissions = default_service_permissions
fetched = json_resp['data'][0]
assert str(sample_user.id) == fetched['id'] assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name'] assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number'] assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address'] assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state'] assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)]) assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user(notify_api, notify_db, notify_db_session, sample_service): def test_get_user(client, sample_service):
""" """
Tests GET endpoint '/<user_id>' to retrieve a single service. Tests GET endpoint '/<user_id>' to retrieve a single service.
""" """
with notify_api.test_request_context(): sample_user = sample_service.users[0]
with notify_api.test_client() as client: header = create_authorization_header()
sample_user = sample_service.users[0] resp = client.get(url_for('user.get_user',
header = create_authorization_header() user_id=sample_user.id),
resp = client.get(url_for('user.get_user', headers=[header])
user_id=sample_user.id), assert resp.status_code == 200
headers=[header]) json_resp = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = default_service_permissions expected_permissions = default_service_permissions
fetched = json_resp['data'] fetched = json_resp['data']
assert str(sample_user.id) == fetched['id'] assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name'] assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number'] assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address'] assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state'] assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)]) assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_post_user(notify_api, notify_db, notify_db_session): def test_post_user(client, notify_db, notify_db_session):
""" """
Tests POST endpoint '/' to create a user. Tests POST endpoint '/' to create a user.
""" """
with notify_api.test_request_context(): assert User.query.count() == 0
with notify_api.test_client() as client: data = {
assert User.query.count() == 0 "name": "Test User",
data = { "email_address": "user@digital.cabinet-office.gov.uk",
"name": "Test User", "password": "password",
"email_address": "user@digital.cabinet-office.gov.uk", "mobile_number": "+447700900986",
"password": "password", "logged_in_at": None,
"mobile_number": "+447700900986", "state": "active",
"logged_in_at": None, "failed_login_count": 0,
"state": "active", "permissions": {}
"failed_login_count": 0, }
"permissions": {} auth_header = create_authorization_header()
} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.post(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.create_user'),
resp = client.post( data=json.dumps(data),
url_for('user.create_user'), headers=headers)
data=json.dumps(data), assert resp.status_code == 201
headers=headers) user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first()
assert resp.status_code == 201 json_resp = json.loads(resp.get_data(as_text=True))
user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first() assert json_resp['data']['email_address'] == user.email_address
json_resp = json.loads(resp.get_data(as_text=True)) assert json_resp['data']['id'] == str(user.id)
assert json_resp['data']['email_address'] == user.email_address
assert json_resp['data']['id'] == str(user.id)
def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_session): def test_post_user_missing_attribute_email(client, notify_db, notify_db_session):
""" """
Tests POST endpoint '/' missing attribute email. Tests POST endpoint '/' missing attribute email.
""" """
with notify_api.test_request_context(): assert User.query.count() == 0
with notify_api.test_client() as client: data = {
assert User.query.count() == 0 "name": "Test User",
data = { "password": "password",
"name": "Test User", "mobile_number": "+447700900986",
"password": "password", "logged_in_at": None,
"mobile_number": "+447700900986", "state": "active",
"logged_in_at": None, "failed_login_count": 0,
"state": "active", "permissions": {}
"failed_login_count": 0, }
"permissions": {} auth_header = create_authorization_header()
} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.post(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.create_user'),
resp = client.post( data=json.dumps(data),
url_for('user.create_user'), headers=headers)
data=json.dumps(data), assert resp.status_code == 400
headers=headers) assert User.query.count() == 0
assert resp.status_code == 400 json_resp = json.loads(resp.get_data(as_text=True))
assert User.query.count() == 0 assert {'email_address': ['Missing data for required field.']} == json_resp['message']
json_resp = json.loads(resp.get_data(as_text=True))
assert {'email_address': ['Missing data for required field.']} == json_resp['message']
def test_create_user_missing_attribute_password(notify_api, notify_db, notify_db_session): def test_create_user_missing_attribute_password(client, notify_db, notify_db_session):
""" """
Tests POST endpoint '/' missing attribute password. Tests POST endpoint '/' missing attribute password.
""" """
with notify_api.test_request_context(): assert User.query.count() == 0
with notify_api.test_client() as client: data = {
assert User.query.count() == 0 "name": "Test User",
data = { "email_address": "user@digital.cabinet-office.gov.uk",
"name": "Test User", "mobile_number": "+447700900986",
"email_address": "user@digital.cabinet-office.gov.uk", "logged_in_at": None,
"mobile_number": "+447700900986", "state": "active",
"logged_in_at": None, "failed_login_count": 0,
"state": "active", "permissions": {}
"failed_login_count": 0, }
"permissions": {} auth_header = create_authorization_header()
} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.post(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.create_user'),
resp = client.post( data=json.dumps(data),
url_for('user.create_user'), headers=headers)
data=json.dumps(data), assert resp.status_code == 400
headers=headers) assert User.query.count() == 0
assert resp.status_code == 400 json_resp = json.loads(resp.get_data(as_text=True))
assert User.query.count() == 0 assert {'password': ['Missing data for required field.']} == json_resp['message']
json_resp = json.loads(resp.get_data(as_text=True))
assert {'password': ['Missing data for required field.']} == json_resp['message']
def test_put_user(notify_api, notify_db, notify_db_session, sample_service): def test_put_user(client, sample_service):
""" """
Tests PUT endpoint '/' to update a user. Tests PUT endpoint '/' to update a user.
""" """
with notify_api.test_request_context(): assert User.query.count() == 1
with notify_api.test_client() as client: sample_user = sample_service.users[0]
assert User.query.count() == 1 sample_user.failed_login_count = 1
sample_user = sample_service.users[0] new_email = 'new@digital.cabinet-office.gov.uk'
new_email = 'new@digital.cabinet-office.gov.uk' data = {
data = { 'name': sample_user.name,
'name': sample_user.name, 'email_address': new_email,
'email_address': new_email, 'mobile_number': sample_user.mobile_number
'mobile_number': sample_user.mobile_number }
} auth_header = create_authorization_header()
auth_header = create_authorization_header() headers = [('Content-Type', 'application/json'), auth_header]
headers = [('Content-Type', 'application/json'), auth_header] resp = client.put(
resp = client.put( url_for('user.update_user', user_id=sample_user.id),
url_for('user.update_user', user_id=sample_user.id), data=json.dumps(data),
data=json.dumps(data), headers=headers)
headers=headers) assert resp.status_code == 200
assert resp.status_code == 200 assert User.query.count() == 1
assert User.query.count() == 1 json_resp = json.loads(resp.get_data(as_text=True))
json_resp = json.loads(resp.get_data(as_text=True)) assert json_resp['data']['email_address'] == new_email
assert json_resp['data']['email_address'] == new_email expected_permissions = default_service_permissions
expected_permissions = default_service_permissions fetched = json_resp['data']
fetched = json_resp['data']
assert str(sample_user.id) == fetched['id'] assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name'] assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number'] assert sample_user.mobile_number == fetched['mobile_number']
assert new_email == fetched['email_address'] assert new_email == fetched['email_address']
assert sample_user.state == fetched['state'] assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)]) assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
# password wasn't updated, so failed_login_count stays the same
assert sample_user.failed_login_count == 1
@pytest.mark.parametrize('user_attribute, user_value', [ @pytest.mark.parametrize('user_attribute, user_value', [
@@ -203,214 +194,169 @@ def test_post_user_attribute(client, sample_user, user_attribute, user_value):
assert json_resp['data'][user_attribute] == user_value assert json_resp['data'][user_attribute] == user_value
def test_put_user_update_password(notify_api, def test_put_user_update_password(client, sample_service):
notify_db,
notify_db_session,
sample_service):
""" """
Tests PUT endpoint '/' to update a user including their password. Tests PUT endpoint '/' to update a user including their password.
""" """
with notify_api.test_request_context(): assert User.query.count() == 1
with notify_api.test_client() as client: sample_user = sample_service.users[0]
assert User.query.count() == 1 new_password = '1234567890'
sample_user = sample_service.users[0] data = {
new_password = '1234567890' 'name': sample_user.name,
data = { 'email_address': sample_user.email_address,
'name': sample_user.name, 'mobile_number': sample_user.mobile_number,
'email_address': sample_user.email_address, 'password': new_password
'mobile_number': sample_user.mobile_number, }
'password': new_password auth_header = create_authorization_header()
} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.put(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.update_user', user_id=sample_user.id),
resp = client.put( data=json.dumps(data),
url_for('user.update_user', user_id=sample_user.id), headers=headers)
data=json.dumps(data), assert resp.status_code == 200
headers=headers) assert User.query.count() == 1
assert resp.status_code == 200 json_resp = json.loads(resp.get_data(as_text=True))
assert User.query.count() == 1 assert json_resp['data']['password_changed_at'] is not None
json_resp = json.loads(resp.get_data(as_text=True)) data = {'password': new_password}
assert json_resp['data']['password_changed_at'] is not None auth_header = create_authorization_header()
data = {'password': new_password} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.post(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.verify_user_password', user_id=str(sample_user.id)),
resp = client.post( data=json.dumps(data),
url_for('user.verify_user_password', user_id=str(sample_user.id)), headers=headers)
data=json.dumps(data), assert resp.status_code == 204
headers=headers)
assert resp.status_code == 204
def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_user, fake_uuid): def test_put_user_not_exists(client, sample_user, fake_uuid):
""" """
Tests PUT endpoint '/' to update a user doesn't exist. Tests PUT endpoint '/' to update a user doesn't exist.
""" """
with notify_api.test_request_context(): assert User.query.count() == 1
with notify_api.test_client() as client: new_email = 'new@digital.cabinet-office.gov.uk'
assert User.query.count() == 1 data = {'email_address': new_email}
new_email = 'new@digital.cabinet-office.gov.uk' auth_header = create_authorization_header()
data = {'email_address': new_email} headers = [('Content-Type', 'application/json'), auth_header]
auth_header = create_authorization_header() resp = client.put(
headers = [('Content-Type', 'application/json'), auth_header] url_for('user.update_user', user_id=fake_uuid),
resp = client.put( data=json.dumps(data),
url_for('user.update_user', user_id=fake_uuid), headers=headers)
data=json.dumps(data), assert resp.status_code == 404
headers=headers) assert User.query.count() == 1
assert resp.status_code == 404 user = User.query.filter_by(id=str(sample_user.id)).first()
assert User.query.count() == 1 json_resp = json.loads(resp.get_data(as_text=True))
user = User.query.filter_by(id=str(sample_user.id)).first() assert json_resp['result'] == "error"
json_resp = json.loads(resp.get_data(as_text=True)) assert json_resp['message'] == 'No result found'
assert json_resp['result'] == "error"
assert json_resp['message'] == 'No result found'
assert user == sample_user assert user == sample_user
assert user.email_address != new_email assert user.email_address != new_email
def test_get_user_by_email(notify_api, notify_db, notify_db_session, sample_service): def test_get_user_by_email(client, sample_service):
sample_user = sample_service.users[0]
header = create_authorization_header()
url = url_for('user.get_by_email', email=sample_user.email_address)
resp = client.get(url, headers=[header])
assert resp.status_code == 200
with notify_api.test_request_context(): json_resp = json.loads(resp.get_data(as_text=True))
with notify_api.test_client() as client: expected_permissions = default_service_permissions
sample_user = sample_service.users[0] fetched = json_resp['data']
header = create_authorization_header()
url = url_for('user.get_by_email', email=sample_user.email_address)
resp = client.get(url, headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True)) assert str(sample_user.id) == fetched['id']
expected_permissions = default_service_permissions assert sample_user.name == fetched['name']
fetched = json_resp['data'] assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert str(sample_user.id) == fetched['id'] assert sample_user.state == fetched['state']
assert sample_user.name == fetched['name'] assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user_by_email_not_found_returns_404(notify_api, def test_get_user_by_email_not_found_returns_404(client, sample_user):
notify_db, header = create_authorization_header()
notify_db_session, url = url_for('user.get_by_email', email='no_user@digital.gov.uk')
sample_user): resp = client.get(url, headers=[header])
assert resp.status_code == 404
with notify_api.test_request_context(): json_resp = json.loads(resp.get_data(as_text=True))
with notify_api.test_client() as client: assert json_resp['result'] == 'error'
header = create_authorization_header() assert json_resp['message'] == 'No result found'
url = url_for('user.get_by_email', email='no_user@digital.gov.uk')
resp = client.get(url, headers=[header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_get_user_by_email_bad_url_returns_404(notify_api, def test_get_user_by_email_bad_url_returns_404(client, sample_user):
notify_db, header = create_authorization_header()
notify_db_session, url = '/user/email'
sample_user): resp = client.get(url, headers=[header])
assert resp.status_code == 400
with notify_api.test_request_context(): json_resp = json.loads(resp.get_data(as_text=True))
with notify_api.test_client() as client: assert json_resp['result'] == 'error'
header = create_authorization_header() assert json_resp['message'] == 'Invalid request. Email query string param required'
url = '/user/email'
resp = client.get(url, headers=[header])
assert resp.status_code == 400
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'Invalid request. Email query string param required'
def test_get_user_with_permissions(notify_api, def test_get_user_with_permissions(client, sample_service_permission):
notify_db, header = create_authorization_header()
notify_db_session, response = client.get(url_for('user.get_user', user_id=str(sample_service_permission.user.id)),
sample_service_permission): headers=[header])
with notify_api.test_request_context(): assert response.status_code == 200
with notify_api.test_client() as client: permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
header = create_authorization_header() assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
response = client.get(url_for('user.get_user', user_id=str(sample_service_permission.user.id)),
headers=[header])
assert response.status_code == 200
permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
def test_set_user_permissions(notify_api, def test_set_user_permissions(client, sample_user, sample_service):
notify_db, data = json.dumps([{'permission': MANAGE_SETTINGS}])
notify_db_session, header = create_authorization_header()
sample_user, headers = [('Content-Type', 'application/json'), header]
sample_service): response = client.post(
with notify_api.test_request_context(): url_for(
with notify_api.test_client() as client: 'user.set_permissions',
data = json.dumps([{'permission': MANAGE_SETTINGS}]) user_id=str(sample_user.id),
header = create_authorization_header() service_id=str(sample_service.id)),
headers = [('Content-Type', 'application/json'), header] headers=headers,
response = client.post( data=data)
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204 assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first() permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user assert permission.user == sample_user
assert permission.service == sample_service assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS assert permission.permission == MANAGE_SETTINGS
def test_set_user_permissions_multiple(notify_api, def test_set_user_permissions_multiple(client, sample_user, sample_service):
notify_db, data = json.dumps([{'permission': MANAGE_SETTINGS}, {'permission': MANAGE_TEMPLATES}])
notify_db_session, header = create_authorization_header()
sample_user, headers = [('Content-Type', 'application/json'), header]
sample_service): response = client.post(
with notify_api.test_request_context(): url_for(
with notify_api.test_client() as client: 'user.set_permissions',
data = json.dumps([{'permission': MANAGE_SETTINGS}, {'permission': MANAGE_TEMPLATES}]) user_id=str(sample_user.id),
header = create_authorization_header() service_id=str(sample_service.id)),
headers = [('Content-Type', 'application/json'), header] headers=headers,
response = client.post( data=data)
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204 assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first() permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user assert permission.user == sample_user
assert permission.service == sample_service assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS assert permission.permission == MANAGE_SETTINGS
permission = Permission.query.filter_by(permission=MANAGE_TEMPLATES).first() permission = Permission.query.filter_by(permission=MANAGE_TEMPLATES).first()
assert permission.user == sample_user assert permission.user == sample_user
assert permission.service == sample_service assert permission.service == sample_service
assert permission.permission == MANAGE_TEMPLATES assert permission.permission == MANAGE_TEMPLATES
def test_set_user_permissions_remove_old(notify_api, def test_set_user_permissions_remove_old(client, sample_user, sample_service):
notify_db, data = json.dumps([{'permission': MANAGE_SETTINGS}])
notify_db_session, header = create_authorization_header()
sample_user, headers = [('Content-Type', 'application/json'), header]
sample_service): response = client.post(
with notify_api.test_request_context(): url_for(
with notify_api.test_client() as client: 'user.set_permissions',
data = json.dumps([{'permission': MANAGE_SETTINGS}]) user_id=str(sample_user.id),
header = create_authorization_header() service_id=str(sample_service.id)),
headers = [('Content-Type', 'application/json'), header] headers=headers,
response = client.post( data=data)
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204 assert response.status_code == 204
query = Permission.query.filter_by(user=sample_user) query = Permission.query.filter_by(user=sample_user)
assert query.count() == 1 assert query.count() == 1
assert query.first().permission == MANAGE_SETTINGS assert query.first().permission == MANAGE_SETTINGS
@freeze_time("2016-01-01 11:09:00.061258") @freeze_time("2016-01-01 11:09:00.061258")
@@ -446,8 +392,7 @@ def test_send_user_reset_password_should_return_400_when_email_is_missing(client
assert mocked.call_count == 0 assert mocked.call_count == 0
def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(client, def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(client, mocker):
mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async') mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
bad_email_address = 'bad@email.gov.uk' bad_email_address = 'bad@email.gov.uk'
data = json.dumps({'email': bad_email_address}) data = json.dumps({'email': bad_email_address})
@@ -495,15 +440,15 @@ def test_send_already_registered_email(client, sample_user, already_registered_t
def test_send_already_registered_email_returns_400_when_data_is_missing(client, sample_user): def test_send_already_registered_email_returns_400_when_data_is_missing(client, sample_user):
data = json.dumps({}) data = json.dumps({})
auth_header = create_authorization_header() auth_header = create_authorization_header()
resp = client.post( resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)), url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data, data=data,
headers=[('Content-Type', 'application/json'), auth_header]) headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400 assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']} assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']}
def test_send_user_confirm_new_email_returns_204(client, sample_user, change_email_confirmation_template, mocker): def test_send_user_confirm_new_email_returns_204(client, sample_user, change_email_confirmation_template, mocker):
@@ -535,7 +480,6 @@ def test_send_user_confirm_new_email_returns_400_when_email_missing(client, samp
def test_update_user_password_saves_correctly(client, sample_service): def test_update_user_password_saves_correctly(client, sample_service):
assert User.query.count() == 1
sample_user = sample_service.users[0] sample_user = sample_service.users[0]
new_password = '1234567890' new_password = '1234567890'
data = { data = {
@@ -548,7 +492,7 @@ def test_update_user_password_saves_correctly(client, sample_service):
data=json.dumps(data), data=json.dumps(data),
headers=headers) headers=headers)
assert resp.status_code == 200 assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True)) json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['password_changed_at'] is not None assert json_resp['data']['password_changed_at'] is not None
data = {'password': new_password} data = {'password': new_password}
@@ -559,3 +503,36 @@ def test_update_user_password_saves_correctly(client, sample_service):
data=json.dumps(data), data=json.dumps(data),
headers=headers) headers=headers)
assert resp.status_code == 204 assert resp.status_code == 204
def test_update_user_password_resets_failed_login_count(client, sample_service):
user = sample_service.users[0]
user.failed_login_count = 1
resp = client.post(
url_for('user.update_password', user_id=user.id),
data=json.dumps({'_password': 'foo'}),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert resp.status_code == 200
assert user.failed_login_count == 0
def test_update_user_resets_failed_login_count_if_updating_password(client, sample_service):
user = sample_service.users[0]
user.failed_login_count = 1
resp = client.put(
url_for('user.update_user', user_id=user.id),
data=json.dumps({
'name': user.name,
'email_address': user.email_address,
'mobile_number': user.mobile_number,
'password': 'foo'
}),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert resp.status_code == 200
assert user.failed_login_count == 0

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

@@ -1,27 +1,24 @@
import json import json
import uuid import uuid
import pytest
from datetime import ( from datetime import (
datetime, datetime,
timedelta timedelta
) )
import pytest
from flask import url_for, current_app from flask import url_for, current_app
from freezegun import freeze_time
from app.dao.services_dao import dao_update_service, dao_fetch_service_by_id from app.dao.services_dao import dao_update_service, dao_fetch_service_by_id
from app.models import ( from app.models import (
VerifyCode, VerifyCode,
User, User,
Notification Notification
) )
from app import db from app import db
import app.celery.tasks
from tests import create_authorization_header from tests import create_authorization_header
from freezegun import freeze_time
import app.celery.tasks
def test_user_verify_code(client, def test_user_verify_code(client,
@@ -163,7 +160,7 @@ def test_user_verify_password_missing_password(client,
@pytest.mark.parametrize('research_mode', [True, False]) @pytest.mark.parametrize('research_mode', [True, False])
@freeze_time("2016-01-01 11:09:00.061258") @freeze_time("2016-01-01 11:09:00.061258")
def test_send_user_sms_code(notify_api, def test_send_user_sms_code(client,
sample_user, sample_user,
sms_code_template, sms_code_template,
mocker, mocker,
@@ -171,68 +168,63 @@ def test_send_user_sms_code(notify_api,
""" """
Tests POST endpoint /user/<user_id>/sms-code Tests POST endpoint /user/<user_id>/sms-code
""" """
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)
with notify_api.test_request_context(): auth_header = create_authorization_header()
with notify_api.test_client() as client: mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
if research_mode: mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notify_service = dao_fetch_service_by_id(current_app.config['NOTIFY_SERVICE_ID'])
notify_service.research_mode = True
dao_update_service(notify_service)
auth_header = create_authorization_header() resp = client.post(
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111') url_for('user.send_user_sms_code', user_id=sample_user.id),
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async') data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
resp = client.post( assert mocked.call_count == 1
url_for('user.send_user_sms_code', user_id=sample_user.id), assert VerifyCode.query.count() == 1
data=json.dumps({}), assert VerifyCode.query.first().check_code('11111')
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert mocked.call_count == 1 assert Notification.query.count() == 1
assert VerifyCode.query.count() == 1 notification = Notification.query.first()
assert VerifyCode.query.first().check_code('11111') assert notification.personalisation == {'verify_code': '11111'}
assert notification.to == sample_user.mobile_number
assert str(notification.service_id) == current_app.config['NOTIFY_SERVICE_ID']
assert Notification.query.count() == 1 app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
notification = Notification.query.first() ([str(notification.id)]),
assert notification.personalisation == {'verify_code': '11111'} queue="notify"
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") @freeze_time("2016-01-01 11:09:00.061258")
def test_send_user_code_for_sms_with_optional_to_field(notify_api, def test_send_user_code_for_sms_with_optional_to_field(client,
sample_user, sample_user,
sms_code_template, sms_code_template,
mocker): mocker):
""" """
Tests POST endpoint /user/<user_id>/sms-code with optional to field Tests POST endpoint /user/<user_id>/sms-code with optional to field
""" """
with notify_api.test_request_context(): to_number = '+441119876757'
with notify_api.test_client() as client: mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
to_number = '+441119876757' mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111') auth_header = create_authorization_header()
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
auth_header = create_authorization_header()
resp = client.post( resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id), url_for('user.send_user_sms_code', user_id=sample_user.id),
data=json.dumps({'to': to_number}), data=json.dumps({'to': to_number}),
headers=[('Content-Type', 'application/json'), auth_header]) headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204 assert resp.status_code == 204
assert mocked.call_count == 1 assert mocked.call_count == 1
notification = Notification.query.first() notification = Notification.query.first()
assert notification.to == to_number assert notification.to == to_number
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with( app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
([str(notification.id)]), ([str(notification.id)]),
queue="notify" queue="notify"
) )
def test_send_sms_code_returns_404_for_bad_input_data(client): def test_send_sms_code_returns_404_for_bad_input_data(client):
@@ -282,12 +274,11 @@ def test_send_user_email_verification(client,
mocked.assert_called_once_with(([str(notification.id)]), queue="notify") mocked.assert_called_once_with(([str(notification.id)]), queue="notify")
def test_send_email_verification_returns_404_for_bad_input_data(client, notify_db, notify_db_session, mocker): def test_send_email_verification_returns_404_for_bad_input_data(client, notify_db_session, mocker):
""" """
Tests POST endpoint /user/<user_id>/sms-code return 404 for bad input data Tests POST endpoint /user/<user_id>/sms-code return 404 for bad input data
""" """
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async') mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
import uuid
uuid_ = uuid.uuid4() uuid_ = uuid.uuid4()
auth_header = create_authorization_header() auth_header = create_authorization_header()
resp = client.post( resp = client.post(
@@ -297,3 +288,17 @@ def test_send_email_verification_returns_404_for_bad_input_data(client, notify_d
assert resp.status_code == 404 assert resp.status_code == 404
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found' assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
assert mocked.call_count == 0 assert mocked.call_count == 0
def test_user_verify_user_code_valid_code_resets_failed_login_count(client, sample_sms_code):
sample_sms_code.user.failed_login_count = 1
data = json.dumps({
'code_type': sample_sms_code.code_type,
'code': sample_sms_code.txt_code})
resp = client.post(
url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert resp.status_code == 204
assert sample_sms_code.user.failed_login_count == 0
assert sample_sms_code.code_used