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

Merge pull request #778 from alphagov/refactor-notify-user-endpoints

Browse Source 
Refactor notify user endpoints
This commit is contained in:
Rebecca Law
2016-12-22 14:50:12 +00:00
committed by GitHub
parent 531cede6cf 7bec9aefcc
commit b2c293e53f
5 changed files with 249 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/invite/rest.py
View File

@@ -1,20 +1,18 @@
import uuid
from datetime import datetime
from flask import (
Blueprint,
request,
jsonify,
current_app)
from app import encryption, DATETIME_FORMAT
from app.dao.invited_user_dao import (
save_invited_user,
get_invited_user,
get_invited_users_for_service
)
from app.dao.templates_dao import dao_get_template_by_id
from app.models import EMAIL_TYPE, KEY_TYPE_NORMAL
from app.notifications.process_notifications import persist_notification, send_notification_to_queue
from app.schemas import invited_user_schema
from app.celery.tasks import (send_email)
invite = Blueprint('invite', __name__, url_prefix='/service/<service_id>/invite')
@@ -29,22 +27,23 @@ def create_invited_user(service_id):
save_invited_user(invited_user)
template = dao_get_template_by_id(current_app.config['INVITATION_EMAIL_TEMPLATE_ID'])
message = {
'template': str(template.id),
'template_version': template.version,
'to': invited_user.email_address,
'personalisation': {
saved_notification = persist_notification(
template_id=template.id,
template_version=template.version,
recipient=invited_user.email_address,
service_id=current_app.config['NOTIFY_SERVICE_ID'],
personalisation={
'user_name': invited_user.from_user.name,
'service_name': invited_user.service.name,
'url': invited_user_url(invited_user.id)
}
}
send_email.apply_async((
current_app.config['NOTIFY_SERVICE_ID'],
str(uuid.uuid4()),
encryption.encrypt(message),
datetime.utcnow().strftime(DATETIME_FORMAT)
), queue="notify")
},
notification_type=EMAIL_TYPE,
api_key_id=None,
key_type=KEY_TYPE_NORMAL
)
send_notification_to_queue(saved_notification, False, queue="notify")
return jsonify(data=invited_user_schema.dump(invited_user).data), 201

58
app/user/rest.py
View File

@@ -1,8 +1,6 @@
import json
import uuid
from datetime import datetime
from flask import (jsonify, request, Blueprint, current_app)
from app import encryption, DATETIME_FORMAT
from app.dao.users_dao import (
get_user_by_id,
save_model_user,
@@ -32,10 +30,6 @@ from app.schemas import (
user_update_schema_load_json
)
from app.celery.tasks import (
send_email
)
from app.errors import (
register_errors,
InvalidRequest
@@ -87,7 +81,6 @@ def update_user_attribute(user_id):
def verify_user_password(user_id):
user_to_verify = get_user_by_id(user_id=user_id)
txt_pwd = None
try:
txt_pwd = request.get_json()['password']
except KeyError:
@@ -227,22 +220,22 @@ def send_already_registered_email(user_id):
to, errors = email_data_request_schema.load(request.get_json())
template = dao_get_template_by_id(current_app.config['ALREADY_REGISTERED_EMAIL_TEMPLATE_ID'])
message = {
'template': str(template.id),
'template_version': template.version,
'to': to['email'],
'personalisation': {
saved_notification = persist_notification(
template_id=template.id,
template_version=template.version,
recipient=to['email'],
service_id=current_app.config['NOTIFY_SERVICE_ID'],
personalisation={
'signin_url': current_app.config['ADMIN_BASE_URL'] + '/sign-in',
'forgot_password_url': current_app.config['ADMIN_BASE_URL'] + '/forgot-password',
'feedback_url': current_app.config['ADMIN_BASE_URL'] + '/feedback'
}
}
send_email.apply_async((
current_app.config['NOTIFY_SERVICE_ID'],
str(uuid.uuid4()),
encryption.encrypt(message),
datetime.utcnow().strftime(DATETIME_FORMAT)
), queue='notify')
},
notification_type=EMAIL_TYPE,
api_key_id=None,
key_type=KEY_TYPE_NORMAL
)
send_notification_to_queue(saved_notification, False, queue="notify")
return jsonify({}), 204
@@ -289,19 +282,22 @@ def send_user_reset_password():
user_to_send_to = get_user_by_email(email['email'])
template = dao_get_template_by_id(current_app.config['PASSWORD_RESET_TEMPLATE_ID'])
message = {
'template': str(template.id),
'template_version': template.version,
'to': user_to_send_to.email_address,
'personalisation': {
saved_notification = persist_notification(
template_id=template.id,
template_version=template.version,
recipient=email['email'],
service_id=current_app.config['NOTIFY_SERVICE_ID'],
personalisation={
'user_name': user_to_send_to.name,
'url': _create_reset_password_url(user_to_send_to.email_address)
}
}
send_email.apply_async([current_app.config['NOTIFY_SERVICE_ID'],
str(uuid.uuid4()),
encryption.encrypt(message),
datetime.utcnow().strftime(DATETIME_FORMAT)], queue='notify')
},
notification_type=EMAIL_TYPE,
api_key_id=None,
key_type=KEY_TYPE_NORMAL
)
send_notification_to_queue(saved_notification, False, queue="notify")
return jsonify({}), 204

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

@@ -1,95 +1,70 @@
import json
import uuid
from flask import current_app
from freezegun import freeze_time
from app import encryption
from app.models import Notification
from tests import create_authorization_header
import app.celery.tasks
@freeze_time("2016-01-01T11:09:00.061258")
def test_create_invited_user(notify_api, sample_service, mocker, invitation_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('uuid.uuid4', return_value='some_uuid') # for the notification id
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('notifications_utils.url_safe_token.generate_token', return_value='the-token')
email_address = 'invited_user@service.gov.uk'
invite_from = sample_service.users[0]
def test_create_invited_user(client, sample_service, mocker, invitation_email_template):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
email_address = 'invited_user@service.gov.uk'
invite_from = sample_service.users[0]
data = {
'service': str(sample_service.id),
'email_address': email_address,
'from_user': str(invite_from.id),
'permissions': 'send_messages,manage_service,manage_api_keys'
}
auth_header = create_authorization_header()
data = {
'service': str(sample_service.id),
'email_address': email_address,
'from_user': str(invite_from.id),
'permissions': 'send_messages,manage_service,manage_api_keys'
}
auth_header = create_authorization_header()
response = client.post(
'/service/{}/invite'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json.dumps(data)
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
response = client.post(
'/service/{}/invite'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json.dumps(data)
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['email_address'] == email_address
assert json_resp['data']['from_user'] == str(invite_from.id)
assert json_resp['data']['permissions'] == 'send_messages,manage_service,manage_api_keys'
assert json_resp['data']['id']
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['email_address'] == email_address
assert json_resp['data']['from_user'] == str(invite_from.id)
assert json_resp['data']['permissions'] == 'send_messages,manage_service,manage_api_keys'
assert json_resp['data']['id']
message = {
'template': str(invitation_email_template.id),
'template_version': invitation_email_template.version,
'to': email_address,
'personalisation': {
'user_name': invite_from.name,
'service_name': sample_service.name,
'url': '{0}/invitation/{1}'.format(current_app.config['ADMIN_BASE_URL'], '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")
notification = Notification.query.first()
mocked.assert_called_once_with([(str(notification.id))], queue="notify")
def test_create_invited_user_invalid_email(notify_api, sample_service, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_email.apply_async')
email_address = 'notanemail'
invite_from = sample_service.users[0]
def test_create_invited_user_invalid_email(client, sample_service, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
email_address = 'notanemail'
invite_from = sample_service.users[0]
data = {
'service': str(sample_service.id),
'email_address': email_address,
'from_user': str(invite_from.id),
'permissions': 'send_messages,manage_service,manage_api_keys'
}
data = {
'service': str(sample_service.id),
'email_address': email_address,
'from_user': str(invite_from.id),
'permissions': 'send_messages,manage_service,manage_api_keys'
}
data = json.dumps(data)
data = json.dumps(data)
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.post(
'/service/{}/invite'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == {'email_address': ['Not a valid email address.']}
app.celery.tasks.send_email.apply_async.assert_not_called()
response = client.post(
'/service/{}/invite'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == {'email_address': ['Not a valid email address.']}
assert mocked.call_count == 0
def test_get_all_invited_users_by_service(notify_api, notify_db, notify_db_session, sample_service):
def test_get_all_invited_users_by_service(client, notify_db, notify_db_session, sample_service):
from tests.app.conftest import sample_invited_user
invites = []
@@ -102,120 +77,101 @@ def test_get_all_invited_users_by_service(notify_api, notify_db, notify_db_sessi
email)
invites.append(invited_user)
with notify_api.test_request_context():
with notify_api.test_client() as client:
url = '/service/{}/invite'.format(sample_service.id)
url = '/service/{}/invite'.format(sample_service.id)
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
invite_from = sample_service.users[0]
invite_from = sample_service.users[0]
for invite in json_resp['data']:
assert invite['service'] == str(sample_service.id)
assert invite['from_user'] == str(invite_from.id)
assert invite['id']
for invite in json_resp['data']:
assert invite['service'] == str(sample_service.id)
assert invite['from_user'] == str(invite_from.id)
assert invite['id']
def test_get_invited_users_by_service_with_no_invites(notify_api, notify_db, notify_db_session, sample_service):
def test_get_invited_users_by_service_with_no_invites(client, notify_db, notify_db_session, sample_service):
url = '/service/{}/invite'.format(sample_service.id)
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
url = '/service/{}/invite'.format(sample_service.id)
auth_header = create_authorization_header()
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_get_invited_user_by_service_and_id(notify_api, sample_service, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
def test_get_invited_user_by_service_and_id(client, sample_service, sample_invited_user):
url = '/service/{}/invite/{}'.format(sample_service.id, sample_invited_user.id)
url = '/service/{}/invite/{}'.format(sample_service.id, sample_invited_user.id)
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
invite_email_address = sample_invited_user.email_address
invite_from = sample_service.users[0]
invite_email_address = sample_invited_user.email_address
invite_from = sample_service.users[0]
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['email_address'] == invite_email_address
assert json_resp['data']['from_user'] == str(invite_from.id)
assert json_resp['data']['id']
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['email_address'] == invite_email_address
assert json_resp['data']['from_user'] == str(invite_from.id)
assert json_resp['data']['id']
def test_get_invited_user_by_service_but_unknown_invite_id_returns_404(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
unknown_id = uuid.uuid4()
url = '/service/{}/invite/{}'.format(sample_service.id, unknown_id)
def test_get_invited_user_by_service_but_unknown_invite_id_returns_404(client, sample_service):
unknown_id = uuid.uuid4()
url = '/service/{}/invite/{}'.format(sample_service.id, unknown_id)
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 404
response = client.get(
url,
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response.status_code == 404
def test_update_invited_user_set_status_to_cancelled(notify_api, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
def test_update_invited_user_set_status_to_cancelled(client, sample_invited_user):
data = {'status': 'cancelled'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url,
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
data = {'status': 'cancelled'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url,
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
assert json_resp['status'] == 'cancelled'
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
assert json_resp['status'] == 'cancelled'
def test_update_invited_user_for_wrong_service_returns_404(notify_api, sample_invited_user, fake_uuid):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {'status': 'cancelled'}
url = '/service/{0}/invite/{1}'.format(fake_uuid, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
json_response = json.loads(response.get_data(as_text=True))['message']
assert json_response == 'No result found'
def test_update_invited_user_for_wrong_service_returns_404(client, sample_invited_user, fake_uuid):
data = {'status': 'cancelled'}
url = '/service/{0}/invite/{1}'.format(fake_uuid, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
json_response = json.loads(response.get_data(as_text=True))['message']
assert json_response == 'No result found'
def test_update_invited_user_for_invalid_data_returns_400(notify_api, sample_invited_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {'status': 'garbage'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400
def test_update_invited_user_for_invalid_data_returns_400(client, sample_invited_user):
data = {'status': 'garbage'}
url = '/service/{0}/invite/{1}'.format(sample_invited_user.service_id, sample_invited_user.id)
auth_header = create_authorization_header()
response = client.post(url, data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400

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

@@ -414,134 +414,98 @@ def test_set_user_permissions_remove_old(notify_api,
@freeze_time("2016-01-01 11:09:00.061258")
def test_send_user_reset_password_should_send_reset_password_link(notify_api,
def test_send_user_reset_password_should_send_reset_password_link(client,
sample_user,
mocker,
password_reset_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('notifications_utils.url_safe_token.generate_token', return_value='the-token')
mocker.patch('uuid.uuid4', return_value='some_uuid') # for the notification id
mocker.patch('app.celery.tasks.send_email.apply_async')
data = json.dumps({'email': sample_user.email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
data = json.dumps({'email': sample_user.email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
message = {
'template': str(password_reset_email_template.id),
'template_version': password_reset_email_template.version,
'to': sample_user.email_address,
'personalisation': {
'user_name': sample_user.name,
'url': current_app.config['ADMIN_BASE_URL'] + '/new-password/' + 'the-token'
}
}
assert resp.status_code == 204
app.celery.tasks.send_email.apply_async.assert_called_once_with(
[str(current_app.config['NOTIFY_SERVICE_ID']),
'some_uuid',
app.encryption.encrypt(message),
"2016-01-01T11:09:00.061258Z"],
queue="notify")
assert resp.status_code == 204
notification = Notification.query.first()
mocked.assert_called_once_with([str(notification.id)], queue="notify")
def test_send_user_reset_password_should_return_400_when_email_is_missing(notify_api):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({})
auth_header = create_authorization_header()
def test_send_user_reset_password_should_return_400_when_email_is_missing(client, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
data = json.dumps({})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']}
assert mocked.call_count == 0
def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(client,
mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
bad_email_address = 'bad@email.gov.uk'
data = json.dumps({'email': bad_email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
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'
assert mocked.call_count == 0
def test_send_user_reset_password_should_return_400_when_data_is_not_email_address(client, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
bad_email_address = 'bad.email.gov.uk'
data = json.dumps({'email': bad_email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Not a valid email address.']}
assert mocked.call_count == 0
def test_send_already_registered_email(client, sample_user, already_registered_template, mocker):
data = json.dumps({'email': sample_user.email_address})
auth_header = create_authorization_header()
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
notification = Notification.query.first()
mocked.assert_called_once_with(([str(notification.id)]), queue="notify")
def test_send_already_registered_email_returns_400_when_data_is_missing(client, sample_user):
data = json.dumps({})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']}
def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(notify_api,
mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
bad_email_address = 'bad@email.gov.uk'
data = json.dumps({'email': bad_email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
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'
def test_send_user_reset_password_should_return_400_when_data_is_not_email_address(notify_api, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
bad_email_address = 'bad.email.gov.uk'
data = json.dumps({'email': bad_email_address})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Not a valid email address.']}
@freeze_time("2016-01-01 11:09:00.061258")
def test_send_already_registered_email(notify_api, sample_user, already_registered_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({'email': sample_user.email_address})
auth_header = create_authorization_header()
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('uuid.uuid4', return_value='some_uuid') # for the notification id
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
message = {
'template': str(already_registered_template.id),
'template_version': already_registered_template.version,
'to': sample_user.email_address,
'personalisation': {
'signin_url': current_app.config['ADMIN_BASE_URL'] + '/sign-in',
'forgot_password_url': current_app.config['ADMIN_BASE_URL'] + '/forgot-password',
'feedback_url': current_app.config['ADMIN_BASE_URL'] + '/feedback'
}
}
app.celery.tasks.send_email.apply_async.assert_called_once_with(
(str(current_app.config['NOTIFY_SERVICE_ID']),
'some_uuid',
app.encryption.encrypt(message),
"2016-01-01T11:09:00.061258Z"),
queue="notify")
def test_send_already_registered_email_returns_400_when_data_is_missing(notify_api, sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps({})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
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):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
new_email = 'new_address@dig.gov.uk'

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

@@ -324,15 +324,14 @@ def test_send_user_email_verification(client,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
notification = Notification.query.first()
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):
def test_send_email_verification_returns_404_for_bad_input_data(client, notify_db, notify_db_session, mocker):
"""
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')
data = json.dumps({})
import uuid
uuid_ = uuid.uuid4()
@@ -343,3 +342,4 @@ def test_send_email_verification_returns_404_for_bad_input_data(client, notify_d
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'
assert mocked.call_count == 0