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

Merge branch 'master' into rate-limit

Browse Source 
Conflicts:
	app/celery/tasks.py
	tests/app/celery/test_tasks.py
This commit is contained in:
Martyn Inglis
2016-03-09 14:16:59 +00:00
parent 4a880af8ef 03f5f01a51
commit 2a13fe4a54
15 changed files with 247 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
tests/app/celery/test_tasks.py
View File

@@ -1,7 +1,13 @@
import uuid
import pytest
from flask import current_app
from app.celery.tasks import (send_sms, send_sms_code, send_email_code, send_email, process_job, email_invited_user)
from app.celery.tasks import (send_sms,
send_sms_code,
send_email_code,
send_email,
process_job,
email_invited_user,
email_reset_password)
from app import (firetext_client, aws_ses_client, encryption, DATETIME_FORMAT)
from app.clients.email.aws_ses import AwsSesClientException
from app.clients.sms.firetext import FiretextClientException
@@ -33,6 +39,8 @@ def test_should_process_sms_job(sample_job, mocker):
process_job(sample_job.id)
s3.get_job_from_s3.assert_called_once_with(sample_job.bucket_name, sample_job.id)
assert encryption.encrypt.call_args[0][0]['to'] == '+441234123123'
assert encryption.encrypt.call_args[0][0]['personalisation'] == {}
tasks.send_sms.apply_async.assert_called_once_with(
(str(sample_job.service_id),
"uuid",
@@ -160,7 +168,7 @@ def test_should_not_create_send_task_for_empty_file(sample_job, mocker):
@freeze_time("2016-01-01 11:09:00.061258")
def test_should_process_email_job(sample_email_job, mocker):
def test_should_process_email_job(sample_email_job, sample_template, mocker):
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('email'))
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
@@ -169,6 +177,8 @@ def test_should_process_email_job(sample_email_job, mocker):
process_job(sample_email_job.id)
s3.get_job_from_s3.assert_called_once_with(sample_email_job.bucket_name, sample_email_job.id)
assert encryption.encrypt.call_args[0][0]['to'] == 'test@test.com'
assert encryption.encrypt.call_args[0][0]['personalisation'] == {}
tasks.send_email.apply_async.assert_called_once_with(
(str(sample_email_job.service_id),
"uuid",
@@ -182,17 +192,22 @@ def test_should_process_email_job(sample_email_job, mocker):
assert job.status == 'finished'
def test_should_process_all_sms_job(sample_job, mocker):
def test_should_process_all_sms_job(sample_job, sample_job_with_placeholdered_template, mocker):
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('multiple_sms'))
mocker.patch('app.celery.tasks.send_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
mocker.patch('app.celery.tasks.create_uuid', return_value="uuid")
process_job(sample_job.id)
process_job(sample_job_with_placeholdered_template.id)
s3.get_job_from_s3.assert_called_once_with(sample_job.bucket_name, sample_job.id)
s3.get_job_from_s3.assert_called_once_with(
sample_job_with_placeholdered_template.bucket_name,
sample_job_with_placeholdered_template.id
)
assert encryption.encrypt.call_args[0][0]['to'] == '+441234123120'
assert encryption.encrypt.call_args[0][0]['personalisation'] == {'name': 'chris'}
tasks.send_sms.apply_async.call_count == 10
job = jobs_dao.dao_get_job_by_id(sample_job.id)
job = jobs_dao.dao_get_job_by_id(sample_job_with_placeholdered_template.id)
assert job.status == 'finished'
@@ -610,3 +625,22 @@ def test_email_invited_user_should_send_email(notify_api, mocker):
invitation['to'],
expected_subject,
expected_content)
def test_email_reset_password_should_send_email(notify_api, mocker):
with notify_api.test_request_context():
reset_password_message = {'to': 'someone@it.gov.uk',
'name': 'Some One',
'reset_password_url': 'bah'}
mocker.patch('app.aws_ses_client.send_email')
mocker.patch('app.encryption.decrypt', return_value=reset_password_message)
encrypted_message = encryption.encrypt(reset_password_message)
email_reset_password(encrypted_message)
message = tasks.password_reset_message(reset_password_message['name'],
reset_password_message['reset_password_url'])
aws_ses_client.send_email(current_app.config['VERIFY_CODE_FROM_EMAIL_ADDRESS'],
reset_password_message['to'],
"Reset password for GOV.UK Notify",
message)

14
tests/app/conftest.py
View File

@@ -221,6 +221,20 @@ def sample_job(notify_db,
return job
@pytest.fixture(scope='function')
def sample_job_with_placeholdered_template(
notify_db,
notify_db_session,
service=None
):
return sample_job(
notify_db,
notify_db_session,
service=service,
template=sample_template_with_placeholders(notify_db, notify_db_session)
)
@pytest.fixture(scope='function')
def sample_email_job(notify_db,
notify_db_session,

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

@@ -90,7 +90,7 @@ def test_create_invited_user_invalid_email(notify_api, sample_service, mocker):
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': ['Invalid email']}
assert json_resp['message'] == {'email_address': ['Not a valid email address']}
app.celery.tasks.email_invited_user.apply_async.assert_not_called()

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

@@ -578,7 +578,7 @@ def test_should_reject_email_notification_with_bad_email(notify_api, sample_emai
app.celery.tasks.send_email.apply_async.assert_not_called()
assert response.status_code == 400
assert data['result'] == 'error'
assert data['message']['to'][0] == 'Invalid email'
assert data['message']['to'][0] == 'Not a valid email address'
def test_should_reject_email_notification_with_template_id_that_cant_be_found(

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

@@ -1,10 +1,12 @@
import json
import uuid
from flask import url_for
import app
from app.models import (User, Permission, MANAGE_SETTINGS, MANAGE_TEMPLATES)
from app.dao.permissions_dao import default_service_permissions
from app import db
from app import db, encryption
from tests import create_authorization_header
@@ -256,7 +258,7 @@ def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_us
user = User.query.filter_by(id=sample_user.id).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == "error"
assert json_resp['message'] == "User not found"
assert json_resp['message'] == "User not found for id: {}".format("9999")
assert user == sample_user
assert user.email_address != new_email
@@ -284,7 +286,7 @@ def test_get_user_by_email(notify_api, notify_db, notify_db_session, sample_serv
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user_by_email_not_found_returns_400(notify_api,
def test_get_user_by_email_not_found_returns_404(notify_api,
notify_db,
notify_db_session,
sample_user):
@@ -297,7 +299,7 @@ def test_get_user_by_email_not_found_returns_400(notify_api,
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'not found'
assert json_resp['message'] == 'User not found for email address'
def test_get_user_by_email_bad_url_returns_404(notify_api,
@@ -426,3 +428,64 @@ def test_set_user_permissions_remove_old(notify_api,
query = Permission.query.filter_by(user=sample_user)
assert query.count() == 1
assert query.first().permission == MANAGE_SETTINGS
def test_send_user_reset_password_should_send_reset_password_link(notify_api,
sample_user,
mocker,
mock_encryption):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.email_reset_password.apply_async')
data = json.dumps({'email': sample_user.email_address})
auth_header = create_authorization_header(
path=url_for('user.send_user_reset_password'),
method='POST',
request_body=data)
resp = client.post(
url_for('user.send_user_reset_password'),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
app.celery.tasks.email_reset_password.apply_async.assert_called_once_with(['something_encrypted'],
queue='email-reset-password')
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(
path=url_for('user.send_user_reset_password'),
method='POST',
request_body=data)
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'] == 'User not found for email address'
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(
path=url_for('user.send_user_reset_password'),
method='POST',
request_body=data)
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']}

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

@@ -10,8 +10,6 @@ from freezegun import freeze_time
def test_user_verify_code_sms(notify_api,
notify_db,
notify_db_session,
sample_sms_code):
"""
Tests POST endpoint '/<user_id>/verify/code'
@@ -35,8 +33,6 @@ def test_user_verify_code_sms(notify_api,
def test_user_verify_code_sms_missing_code(notify_api,
notify_db,
notify_db_session,
sample_sms_code):
"""
Tests POST endpoint '/<user_id>/verify/code'
@@ -59,8 +55,6 @@ def test_user_verify_code_sms_missing_code(notify_api,
@moto.mock_sqs
def test_user_verify_code_email(notify_api,
notify_db,
notify_db_session,
sqs_client_conn,
sample_email_code):
"""
@@ -85,8 +79,6 @@ def test_user_verify_code_email(notify_api,
def test_user_verify_code_email_bad_code(notify_api,
notify_db,
notify_db_session,
sample_email_code):
"""
Tests POST endpoint '/<user_id>/verify/code'
@@ -110,8 +102,6 @@ def test_user_verify_code_email_bad_code(notify_api,
def test_user_verify_code_email_expired_code(notify_api,
notify_db,
notify_db_session,
sample_email_code):
"""
Tests POST endpoint '/<user_id>/verify/code'
@@ -162,8 +152,6 @@ def test_user_verify_password(notify_api,
def test_user_verify_password_invalid_password(notify_api,
notify_db,
notify_db_session,
sample_user):
"""
Tests POST endpoint '/<user_id>/verify/password' invalid endpoint.
@@ -189,8 +177,6 @@ def test_user_verify_password_invalid_password(notify_api,
def test_user_verify_password_valid_password_resets_failed_logins(notify_api,
notify_db,
notify_db_session,
sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -227,8 +213,6 @@ def test_user_verify_password_valid_password_resets_failed_logins(notify_api,
def test_user_verify_password_missing_password(notify_api,
notify_db,
notify_db_session,
sample_user):
"""
Tests POST endpoint '/<user_id>/verify/password' missing password.
@@ -314,7 +298,7 @@ def test_send_sms_code_returns_404_for_bad_input_data(notify_api, notify_db, not
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 user found'
assert json.loads(resp.get_data(as_text=True))['message'] == 'User not found for id: {}'.format(int(uuid_))
def test_send_user_email_code(notify_api,
@@ -356,4 +340,4 @@ def test_send_user_email_code_returns_404_for_when_user_does_not_exist(notify_ap
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 user found'
assert json.loads(resp.get_data(as_text=True))['message'] == 'User not found for id: {}'.format(1)