darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-19 22:14:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #32 from alphagov/add_verify_code_rest

Browse Source 
All tests working, second time around.
This commit is contained in:
NIcholas Staples
2016-01-22 15:00:08 +00:00
parent 02ffbb4f67 e1c8360fd1
commit 2c9ac187c6
15 changed files with 601 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/dao/templates_dao.py
View File

@@ -23,6 +23,7 @@ def delete_model_template(template):
def get_model_templates(template_id=None, service_id=None):
temp = Template.query.first()
# TODO need better mapping from function params to sql query.
if template_id and service_id:
return Template.query.filter_by(

41
app/dao/users_dao.py
View File

@@ -1,9 +1,15 @@
from datetime import datetime
import random
from datetime import (datetime, timedelta)
from . import DAOException
from sqlalchemy.orm import load_only
from app import db
from app.models import User
from app.models import (User, VerifyCode)
from app.encryption import hashpw
def create_secret_code():
return ''.join(map(str, random.sample(range(9), 5)))
def save_model_user(usr, update_dict={}):
@@ -16,6 +22,37 @@ def save_model_user(usr, update_dict={}):
db.session.commit()
def create_user_code(user, code, code_type):
verify_code = VerifyCode(code_type=code_type,
expiry_datetime=datetime.now() + timedelta(hours=1),
user=user)
verify_code.code = code
db.session.add(verify_code)
db.session.commit()
return verify_code
def get_user_code(user, code, code_type):
# Get the most recent codes to try and reduce the
# time searching for the correct code.
codes = VerifyCode.query.filter_by(
user=user, code_type=code_type).order_by(
VerifyCode.created_at.desc())
retval = None
for x in codes:
if x.check_code(code):
retval = x
break
return retval
def use_user_code(id):
verify_code = VerifyCode.query.get(id)
verify_code.code_used = True
db.session.add(verify_code)
db.session.commit()
def delete_model_user(user):
db.session.delete(user)
db.session.commit()

31
app/models.py
View File

@@ -150,3 +150,34 @@ class Job(db.Model):
unique=False,
nullable=True,
onupdate=datetime.datetime.now)
class VerifyCode(db.Model):
__tablename__ = 'verify_codes'
code_types = ['email', 'sms']
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), index=True, nullable=False)
user = db.relationship('User', backref=db.backref('verify_codes', lazy='dynamic'))
_code = db.Column(db.String, nullable=False)
code_type = db.Column(db.Enum(*code_types, name='verify_code_types'), index=False, unique=False, nullable=False)
expiry_datetime = db.Column(db.DateTime, nullable=False)
code_used = db.Column(db.Boolean, default=False)
created_at = db.Column(
db.DateTime,
index=False,
unique=False,
nullable=False,
default=datetime.datetime.now)
@property
def code(self):
raise AttributeError("Code not readable")
@code.setter
def code(self, cde):
self._code = hashpw(cde)
def check_code(self, cde):
return check_hash(cde, self._code)

84
app/notifications/rest.py
View File

@@ -1,10 +1,13 @@
from flask import (
Blueprint,
jsonify,
request
request,
current_app
)
from app import notify_alpha_client
from app import api_user
from app.dao import (templates_dao, services_dao)
import re
mobile_regex = re.compile("^\\+44[\\d]{10}$")
@@ -21,19 +24,19 @@ def get_notifications(notification_id):
def create_sms_notification():
notification = request.get_json()['notification']
errors = {}
to_errors = validate_to(notification)
message_errors = validate_message(notification)
if to_errors:
to, to_errors = validate_to(notification, api_user['client'])
print("create sms")
print(notification)
template, template_errors = validate_template(notification, api_user['client'])
if to_errors['to']:
errors.update(to_errors)
if message_errors:
errors.update(message_errors)
if template_errors['template']:
errors.update(template_errors)
if errors:
return jsonify(result="error", message=errors), 400
return jsonify(notify_alpha_client.send_sms(
mobile_number=notification['to'],
message=notification['message'])), 200
mobile_number=to,
message=template)), 200
@notifications.route('/email', methods=['POST'])
@@ -55,36 +58,45 @@ def create_email_notification():
notification['subject']))
def validate_to(json_body):
errors = []
if 'to' not in json_body:
errors.append('required')
def validate_to(json_body, service_id):
errors = {"to": []}
mob = json_body.get('to', None)
if not mob:
errors['to'].append('Required data missing')
else:
if not mobile_regex.match(json_body['to']):
errors.append('invalid phone number, must be of format +441234123123')
if errors:
return {
"to": errors
}
return None
if not mobile_regex.match(mob):
errors['to'].append('invalid phone number, must be of format +441234123123')
if service_id != current_app.config.get('ADMIN_CLIENT_USER_NAME'):
service = services_dao.get_model_services(service_id=service_id)
if service.restricted:
valid = False
for usr in service.users:
if mob == usr.mobile_number:
valid = True
break
if not valid:
errors['to'].append('Invalid phone number for restricted service')
return mob, errors
def validate_message(json_body):
errors = []
if 'message' not in json_body:
errors.append('required')
def validate_template(json_body, service_id):
errors = {"template": []}
template_id = json_body.get('template', None)
content = ''
if not template_id:
errors['template'].append('Required data missing')
else:
message_length = len(json_body['message'])
if message_length < 1 or message_length > 160:
errors.append('Invalid length. [1 - 160]')
if errors:
return {
"message": errors
}
return None
if service_id == current_app.config.get('ADMIN_CLIENT_USER_NAME'):
content = json_body['template']
else:
try:
template = templates_dao.get_model_templates(
template_id=json_body['template'],
service_id=service_id)
content = template.content
except:
errors['template'].append("Unable to load template.")
return content, errors
def validate_required_and_something(json_body, field):

11
app/schemas.py
View File

@@ -10,7 +10,9 @@ from . import models
class UserSchema(ma.ModelSchema):
class Meta:
model = models.User
exclude = ("updated_at", "created_at", "user_to_service", "_password")
exclude = (
"updated_at", "created_at", "user_to_service",
"_password", "verify_codes")
# TODO process users list, to return a list of user.id
@@ -38,6 +40,12 @@ class JobSchema(ma.ModelSchema):
model = models.Job
class VerifyCodeSchema(ma.ModelSchema):
class Meta:
model = models.VerifyCode
exclude = ('user', "_code", "expiry_datetime", "code_used", "created_at")
user_schema = UserSchema()
users_schema = UserSchema(many=True)
service_schema = ServiceSchema()
@@ -48,3 +56,4 @@ api_key_schema = ApiKeySchema()
api_keys_schema = ApiKeySchema(many=True)
job_schema = JobSchema()
jobs_schema = JobSchema(many=True)
verify_code_schema = VerifyCodeSchema()

84
app/user/rest.py
View File

@@ -1,15 +1,19 @@
from flask import (jsonify, request)
from datetime import datetime
from flask import (jsonify, request, abort)
from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound
from app.dao.services_dao import get_model_services
from app.dao.users_dao import (
get_model_users, save_model_user, delete_model_user)
get_model_users, save_model_user, delete_model_user,
create_user_code, get_user_code, use_user_code,
create_secret_code)
from app.schemas import (
user_schema, users_schema, service_schema, services_schema)
user_schema, users_schema, service_schema, services_schema,
verify_code_schema)
from app import db
from flask import Blueprint
user = Blueprint('user', __name__)
@@ -57,19 +61,81 @@ def verify_user_password(user_id):
return jsonify(result="error", message="Invalid user id"), 400
except NoResultFound:
return jsonify(result="error", message="User not found"), 404
text_pwd = None
txt_pwd = None
try:
text_pwd = request.get_json()['password']
txt_pwd = request.get_json()['password']
except KeyError:
return jsonify(
result="error",
message={'password': ['Required field missing data']}), 400
if user.check_password(text_pwd):
return jsonify(), 204
if user.check_password(txt_pwd):
return jsonify(''), 204
else:
return jsonify(result='error', message={'password': ['Incorrect password']}), 400
@user.route('/<int:user_id>/verify/code', methods=['POST'])
def verify_user_code(user_id):
try:
user = get_model_users(user_id=user_id)
except DataError:
return jsonify(result="error", message="Invalid user id"), 400
except NoResultFound:
return jsonify(result="error", message="User not found"), 404
txt_code = None
resp_json = request.get_json()
txt_type = None
errors = {}
try:
txt_code = resp_json['code']
except KeyError:
errors.update({'code': ['Required field missing data']})
try:
txt_type = resp_json['code_type']
except KeyError:
errors.update({'code_type': ['Required field missing data']})
if errors:
return jsonify(result="error", message=errors), 400
code = get_user_code(user, txt_code, txt_type)
if not code:
return jsonify(result="error", message="Code not found"), 404
if datetime.now() > code.expiry_datetime or code.code_used:
return jsonify(result="error", message="Code has expired"), 400
use_user_code(code.id)
return jsonify(''), 204
@user.route('/<int:user_id>/code/', methods=['POST'])
def send_user_code(user_id):
try:
user = get_model_users(user_id=user_id)
except DataError:
return jsonify(result="error", message="Invalid user id"), 400
except NoResultFound:
return jsonify(result="error", message="User not found"), 404
text_pwd = None
verify_code, errors = verify_code_schema.load(request.get_json())
if errors:
return jsonify(result="error", message=errors), 400
code = create_user_code(
user, create_secret_code(), verify_code.code_type)
# TODO this will need to fixed up when we stop using
# notify_alpha_client
if verify_code.code_type == 'sms':
notify_alpha_client.send_sms(
mobile_number=user.mobile_number,
message=code.code)
elif verify_code.code_type == 'email':
notify_alpha_client.send_email(
user.email_address,
code.code,
'notify@digital.cabinet-office.gov.uk',
'Verification code')
else:
abort(500)
return jsonify(''), 204
@user.route('/<int:user_id>', methods=['GET'])
@user.route('', methods=['GET'])
def get_user(user_id=None):

38
migrations/versions/0008_add_verify_codes.py Normal file
View File

@@ -0,0 +1,38 @@
"""empty message
Revision ID: 0008_add_verify_codes
Revises: 0007_change_to_api_keys
Create Date: 2016-01-21 16:59:05.818017
"""
# revision identifiers, used by Alembic.
revision = '0008_add_verify_codes'
down_revision = '0007_change_to_api_keys'
from alembic import op
import sqlalchemy as sa
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('verify_codes',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('_code', sa.String(), nullable=False),
sa.Column('code_type', sa.Enum('email', 'sms', name='verify_code_types'), nullable=False),
sa.Column('expiry_datetime', sa.DateTime(), nullable=False),
sa.Column('code_used', sa.Boolean(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_verify_codes_user_id'), 'verify_codes', ['user_id'], unique=False)
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_verify_codes_user_id'), table_name='verify_codes')
op.drop_table('verify_codes')
### end Alembic commands ###

6
migrations/versions/0008_unique_service_to_key_name.py → migrations/versions/0009_unique_service_to_key_name.py
View File

@@ -1,14 +1,14 @@
"""empty message
Revision ID: 0008_unique_service_to_key_name
Revision ID: 0009_unique_service_to_key_name
Revises: 0007_change_to_api_keys
Create Date: 2016-01-21 16:14:51.773001
"""
# revision identifiers, used by Alembic.
revision = '0008_unique_service_to_key_name'
down_revision = '0007_change_to_api_keys'
revision = '0009_unique_service_to_key_name'
down_revision = '0008_add_verify_codes'
from alembic import op
import sqlalchemy as sa

2
requirements.txt
View File

@@ -13,7 +13,7 @@ itsdangerous==0.24
Flask-Bcrypt==0.6.2
credstash==1.8.0
git+https://github.com/alphagov/notifications-python-client.git@0.1.9#egg=notifications-python-client==0.1.9
git+https://github.com/alphagov/notifications-python-client.git@0.2.1#egg=notifications-python-client==0.2.1
git+https://github.com/alphagov/notifications-utils.git@0.0.3#egg=notifications-utils==0.0.3

58
tests/app/conftest.py
View File

@@ -1,6 +1,6 @@
import pytest
from app.models import (User, Service, Template, ApiKey, Job)
from app.dao.users_dao import (save_model_user)
from app.models import (User, Service, Template, ApiKey, Job, VerifyCode)
from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
from app.dao.services_dao import save_model_service
from app.dao.templates_dao import save_model_template
from app.dao.api_key_dao import save_model_api_key
@@ -16,12 +16,52 @@ def sample_user(notify_db,
'name': 'Test User',
'email_address': email,
'password': 'password',
'mobile_number': '+44 7700 900986',
'mobile_number': '+447700900986',
'state': 'active'
}
user = User(**data)
save_model_user(user)
return user
usr = User.query.filter_by(email_address=email).first()
if not usr:
usr = User(**data)
save_model_user(usr)
return usr
def create_code(notify_db, notify_db_session, code_type, usr=None, code=None):
if code is None:
code = create_secret_code()
if usr is None:
usr = sample_user(notify_db, notify_db_session)
return create_user_code(usr, code, code_type), code
@pytest.fixture(scope='function')
def sample_email_code(notify_db,
notify_db_session,
code=None,
code_type="email",
usr=None):
code, txt_code = create_code(notify_db,
notify_db_session,
code_type,
usr=usr,
code=code)
code.txt_code = txt_code
return code
@pytest.fixture(scope='function')
def sample_sms_code(notify_db,
notify_db_session,
code=None,
code_type="sms",
usr=None):
code, txt_code = create_code(notify_db,
notify_db_session,
code_type,
usr=usr,
code=code)
code.txt_code = txt_code
return code
@pytest.fixture(scope='function')
@@ -37,8 +77,10 @@ def sample_service(notify_db,
'limit': 1000,
'active': False,
'restricted': False}
service = Service(**data)
save_model_service(service)
service = Service.query.filter_by(name=service_name).first()
if not service:
service = Service(**data)
save_model_service(service)
return service

2
tests/app/dao/test_jobs_dao.py
View File

@@ -52,7 +52,7 @@ def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
other_user = create_user(notify_db, notify_db_session,
email="test@digital.cabinet-office.gov.uk")
other_service = create_service(notify_db, notify_db_session,
user=other_user)
user=other_user, service_name="other service")
other_template = create_template(notify_db, notify_db_session,
service=other_service)
other_job = create_job(notify_db, notify_db_session, service=other_service,

2
tests/app/dao/test_users_dao.py
View File

@@ -12,7 +12,7 @@ def test_create_user(notify_api, notify_db, notify_db_session):
'name': 'Test User',
'email_address': email,
'password': 'password',
'mobile_number': '+44 7700 900986'
'mobile_number': '+447700900986'
}
user = User(**data)
save_model_user(user)

145
tests/app/notifications/test_rest.py
View File

@@ -1,6 +1,7 @@
from tests import create_authorization_header
from flask import url_for, json
from app import notify_alpha_client
from app.models import Service
def test_get_notifications(
@@ -82,7 +83,7 @@ def test_should_reject_if_no_phone_numbers(
)
data = {
'notification': {
'message': "my message"
'template': "my message"
}
}
auth_header = create_authorization_header(
@@ -97,12 +98,9 @@ def test_should_reject_if_no_phone_numbers(
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
print(json_resp)
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert len(json_resp['message']) == 1
assert len(json_resp['message']['to']) == 1
assert json_resp['message']['to'][0] == 'required'
assert 'Required data missing' in json_resp['message']['to'][0]
assert not notify_alpha_client.send_sms.called
@@ -120,7 +118,7 @@ def test_should_reject_bad_phone_numbers(
data = {
'notification': {
'to': 'invalid',
'message': "my message"
'template': "my message"
}
}
auth_header = create_authorization_header(
@@ -135,16 +133,13 @@ def test_should_reject_bad_phone_numbers(
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
print(json_resp)
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert len(json_resp['message']) == 1
assert len(json_resp['message']['to']) == 1
assert json_resp['message']['to'][0] == 'invalid phone number, must be of format +441234123123'
assert 'invalid phone number, must be of format +441234123123' in json_resp['message']['to']
assert not notify_alpha_client.send_sms.called
def test_should_reject_missing_message(
def test_should_reject_missing_template(
notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
@@ -174,31 +169,90 @@ def test_should_reject_missing_message(
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert len(json_resp['message']) == 1
assert len(json_resp['message']['message']) == 1
assert json_resp['message']['message'][0] == 'required'
assert 'Required data missing' in json_resp['message']['template']
assert not notify_alpha_client.send_sms.called
def test_should_reject_too_short_message(
notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id, mocker):
def test_send_template_content(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
Test POST endpoint '/sms' with service notification.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch(
'app.notify_alpha_client.send_sms',
return_value='success'
return_value={
"notification": {
"createdAt": "2015-11-03T09:37:27.414363Z",
"id": 100,
"jobId": 65,
"message": sample_template.content,
"method": "sms",
"status": "created",
"to": sample_user.mobile_number
}
}
)
data = {
'notification': {
'to': '+441234123123',
'message': ''
'to': sample_user.mobile_number,
'template': sample_template.id
}
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert json_resp['notification']['id'] == 100
notify_alpha_client.send_sms.assert_called_with(
mobile_number=sample_user.mobile_number,
message=sample_template.content)
def test_send_notification_restrict_mobile(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Test POST endpoint '/sms' with service notification with mobile number
not in restricted list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
Service.query.filter_by(
id=sample_template.service.id).update({'restricted': True})
invalid_mob = '+449999999999'
mocker.patch(
'app.notify_alpha_client.send_sms',
return_value={}
)
data = {
'notification': {
'to': invalid_mob,
'template': sample_template.id
}
}
assert invalid_mob != sample_user.mobile_number
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
@@ -210,54 +264,13 @@ def test_should_reject_too_short_message(
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert len(json_resp['message']) == 1
assert len(json_resp['message']['message']) == 1
assert json_resp['message']['message'][0] == 'Invalid length. [1 - 160]'
assert not notify_alpha_client.send_sms.called
def test_should_reject_too_long_message(
notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch(
'app.notify_alpha_client.send_sms',
return_value='success'
)
data = {
'notification': {
'to': '+441234123123',
'message': '1' * 161
}
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert len(json_resp['message']) == 1
assert len(json_resp['message']['message']) == 1
assert json_resp['message']['message'][0] == 'Invalid length. [1 - 160]'
assert not notify_alpha_client.send_sms.called
assert 'Invalid phone number for restricted service' in json_resp['message']['to']
def test_should_allow_valid_message(
notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
Tests POST endpoint '/sms' with notifications-admin notification.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -278,11 +291,10 @@ def test_should_allow_valid_message(
data = {
'notification': {
'to': '+441234123123',
'message': 'valid'
'template': 'valid'
}
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
@@ -336,7 +348,6 @@ def test_send_email_valid_data(notify_api,
}
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
request_body=json.dumps(data),
path=url_for('notifications.create_email_notification'),
method='POST')

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

@@ -23,7 +23,7 @@ def test_get_user_list(notify_api, notify_db, notify_db_session, sample_user, sa
"name": "Test User",
"email_address": sample_user.email_address,
"id": sample_user.id,
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"logged_in_at": None,
"state": "active",
@@ -50,7 +50,7 @@ def test_get_user(notify_api, notify_db, notify_db_session, sample_user, sample_
"name": "Test User",
"email_address": sample_user.email_address,
"id": sample_user.id,
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"logged_in_at": None,
"state": "active",
@@ -70,7 +70,7 @@ def test_post_user(notify_api, notify_db, notify_db_session, sample_admin_servic
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"password": "password",
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"logged_in_at": None,
"state": "active",
@@ -103,7 +103,7 @@ def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_sess
data = {
"name": "Test User",
"password": "password",
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"logged_in_at": None,
"state": "active",
@@ -134,7 +134,7 @@ def test_post_user_missing_attribute_password(notify_api, notify_db, notify_db_s
data = {
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"logged_in_at": None,
"state": "active",
@@ -182,7 +182,7 @@ def test_put_user(notify_api, notify_db, notify_db_session, sample_user, sample_
expected = {
"name": "Test User",
"email_address": new_email,
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"id": user.id,
"logged_in_at": None,
@@ -203,12 +203,12 @@ def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_us
new_email = 'new@digital.cabinet-office.gov.uk'
data = {'email_address': new_email}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id="123"),
path=url_for('user.update_user', user_id="9999"),
method='PUT',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id="123"),
url_for('user.update_user', user_id="9999"),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 404
@@ -333,7 +333,7 @@ def test_delete_user(notify_api, notify_db, notify_db_session, sample_user, samp
expected = {
"name": "Test User",
"email_address": sample_user.email_address,
"mobile_number": "+44 7700 900986",
"mobile_number": "+447700900986",
"password_changed_at": None,
"id": sample_user.id,
"logged_in_at": None,
@@ -359,76 +359,3 @@ 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
def test_user_verify_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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
def test_user_verify_password_invalid_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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']
def test_user_verify_password_missing_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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']

216
tests/app/user/test_rest_verify.py Normal file
View File

@@ -0,0 +1,216 @@
import json
from datetime import (datetime, timedelta)
from flask import url_for
from app.models import (User, Service, VerifyCode)
from app import db
from tests import create_authorization_header
def test_user_verify_code_sms(notify_api,
notify_db,
notify_db_session,
sample_admin_service_id,
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
method='POST',
request_body=data)
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,
notify_db,
notify_db_session,
sample_admin_service_id,
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
method='POST',
request_body=data)
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
def test_user_verify_code_email(notify_api,
notify_db,
notify_db_session,
sample_admin_service_id,
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_code', user_id=sample_email_code.user.id),
method='POST',
request_body=data)
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,
notify_db,
notify_db_session,
sample_admin_service_id,
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_code', user_id=sample_email_code.user.id),
method='POST',
request_body=data)
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,
notify_db,
notify_db_session,
sample_admin_service_id,
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.now() - 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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_code', user_id=sample_email_code.user.id),
method='POST',
request_body=data)
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
def test_user_verify_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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
def test_user_verify_password_invalid_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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']
def test_user_verify_password_missing_password(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_admin_service_id):
"""
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(
service_id=sample_admin_service_id,
path=url_for('user.verify_user_password', user_id=sample_user.id),
method='POST',
request_body=data)
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']