darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-28 03:39:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

merged master and up migration version

Browse Source
This commit is contained in:
venusbb
2018-03-16 10:55:36 +00:00
parent bb95a2784f 62d2b04e0a
commit 7e2947790f
30 changed files with 658 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tests/app/accept_invite/test_accept_invite_rest.py
View File

@@ -61,3 +61,23 @@ def test_validate_invitation_token_returns_400_when_invited_user_does_not_exist(
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
@pytest.mark.parametrize('invitation_type', ['service', 'organisation'])
def test_validate_invitation_token_returns_400_when_token_is_malformed(client, invitation_type):
token = generate_token(
str(uuid.uuid4()),
current_app.config['SECRET_KEY'],
current_app.config['DANGEROUS_SALT']
)[:-2]
url = '/invite/{}/{}'.format(invitation_type, token)
auth_header = create_authorization_header()
response = client.get(url, headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == {
'invitation': 'Something’s wrong with this link. Make sure you’ve copied the whole thing.'
}

53
tests/app/celery/test_ftp_update_tasks.py
View File

@@ -59,8 +59,8 @@ def test_update_letter_notifications_statuses_raises_for_invalid_format(notify_a
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=invalid_file)
with pytest.raises(DVLAException) as e:
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
assert 'DVLA response file: {} has an invalid format'.format('NOTIFY.20170823160812.RSP.TXT') in str(e)
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
assert 'DVLA response file: {} has an invalid format'.format('NOTIFY-20170823160812-RSP.TXT') in str(e)
def test_update_letter_notification_statuses_when_notification_does_not_exist_updates_notification_history(
@@ -73,7 +73,7 @@ def test_update_letter_notification_statuses_when_notification_does_not_exist_up
billable_units=1)
Notification.query.filter_by(id=notification.id).delete()
update_letter_notifications_statuses(filename="NOTIFY.20170823160812.RSP.TXT")
update_letter_notifications_statuses(filename="NOTIFY-20170823160812-RSP.TXT")
updated_history = NotificationHistory.query.filter_by(id=notification.id).one()
assert updated_history.status == NOTIFICATION_DELIVERED
@@ -106,10 +106,29 @@ def test_update_letter_notifications_statuses_raises_error_for_unknown_sorted_st
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
with pytest.raises(DVLAException) as e:
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
assert "DVLA response file: {filename} contains unknown Sorted status {unknown_status}".format(
filename="NOTIFY.20170823160812.RSP.TXT", unknown_status="{'Error'}"
filename="NOTIFY-20170823160812-RSP.TXT", unknown_status="{'Error'}"
) in str(e)
def test_update_letter_notifications_statuses_still_raises_temp_failure_error_with_unknown_sorted_status(
notify_api,
mocker,
sample_letter_template
):
valid_file = 'ref-foo|Failed|1|unknown'
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
create_notification(sample_letter_template, reference='ref-foo', status=NOTIFICATION_SENDING,
billable_units=0)
with pytest.raises(DVLAException) as e:
update_letter_notifications_statuses(filename="failed.txt")
failed = ["ref-foo"]
assert "DVLA response file: {filename} has failed letters with notification.reference {failures}".format(
filename="failed.txt", failures=failed
) in str(e)
@@ -117,10 +136,10 @@ def test_update_letter_notifications_statuses_calls_with_correct_bucket_location
s3_mock = mocker.patch('app.celery.tasks.s3.get_s3_object')
with set_config(notify_api, 'NOTIFY_EMAIL_DOMAIN', 'foo.bar'):
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
s3_mock.assert_called_with('{}-ftp'.format(
current_app.config['NOTIFY_EMAIL_DOMAIN']),
'NOTIFY.20170823160812.RSP.TXT'
'NOTIFY-20170823160812-RSP.TXT'
)
@@ -129,7 +148,7 @@ def test_update_letter_notifications_statuses_builds_updates_from_content(notify
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
update_mock = mocker.patch('app.celery.tasks.process_updates_from_file')
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
update_mock.assert_called_with('ref-foo|Sent|1|Unsorted\nref-bar|Sent|2|Sorted')
@@ -162,7 +181,7 @@ def test_update_letter_notifications_statuses_persisted(notify_api, mocker, samp
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
with pytest.raises(expected_exception=DVLAException) as e:
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
assert sent_letter.status == NOTIFICATION_DELIVERED
assert sent_letter.billable_units == 1
@@ -171,7 +190,7 @@ def test_update_letter_notifications_statuses_persisted(notify_api, mocker, samp
assert failed_letter.billable_units == 2
assert failed_letter.updated_at
assert "DVLA response file: {filename} has failed letters with notification.reference {failures}".format(
filename="NOTIFY.20170823160812.RSP.TXT", failures=[format(failed_letter.reference)]) in str(e)
filename="NOTIFY-20170823160812-RSP.TXT", failures=[format(failed_letter.reference)]) in str(e)
def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count(
@@ -187,9 +206,11 @@ def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
persist_letter_count_mock = mocker.patch('app.celery.tasks.persist_daily_sorted_letter_counts')
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
persist_letter_count_mock.assert_called_once_with(date(2017, 8, 23), {'Unsorted': 1, 'Sorted': 1})
persist_letter_count_mock.assert_called_once_with(day=date(2017, 8, 23),
file_name='NOTIFY-20170823160812-RSP.TXT',
sorted_letter_counts={'Unsorted': 1, 'Sorted': 1})
def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count_with_no_sorted_values(
@@ -204,7 +225,7 @@ def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count
sent_letter_1.reference, sent_letter_2.reference)
mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(date(2017, 8, 23))
@@ -223,7 +244,7 @@ def test_update_letter_notifications_does_not_call_send_callback_if_no_db_entry(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
update_letter_notifications_statuses(filename='NOTIFY-20170823160812-RSP.TXT')
send_mock.assert_not_called()
@@ -301,7 +322,7 @@ def test_check_billable_units_when_billable_units_does_not_match_page_count(
('20170120230000', date(2017, 1, 20))
])
def test_get_billing_date_in_bst_from_filename(filename_date, billing_date):
filename = 'NOTIFY.{}.RSP.TXT'.format(filename_date)
filename = 'NOTIFY-{}-RSP.TXT'.format(filename_date)
result = get_billing_date_in_bst_from_filename(filename)
assert result == billing_date
@@ -310,7 +331,7 @@ def test_get_billing_date_in_bst_from_filename(filename_date, billing_date):
@freeze_time("2018-01-11 09:00:00")
def test_persist_daily_sorted_letter_counts_saves_sorted_and_unsorted_values(client, notify_db_session):
letter_counts = defaultdict(int, **{'Unsorted': 5, 'Sorted': 1})
persist_daily_sorted_letter_counts(date.today(), letter_counts)
persist_daily_sorted_letter_counts(date.today(), "test.txt", letter_counts)
day = dao_get_daily_sorted_letter_by_billing_day(date.today())
assert day.unsorted_count == 5

19
tests/app/celery/test_letters_pdf_tasks.py
View File

@@ -28,25 +28,6 @@ def test_should_have_decorated_tasks_functions():
assert create_letters_pdf.__wrapped__.__name__ == 'create_letters_pdf'
@pytest.mark.parametrize('crown_flag,expected_crown_text', [
(True, 'C'),
(False, 'N'),
])
@freeze_time("2017-12-04 17:29:00")
def test_get_letter_pdf_filename_returns_correct_filename(
notify_api, mocker, crown_flag, expected_crown_text):
filename = get_letter_pdf_filename(reference='foo', crown=crown_flag)
assert filename == '2017-12-04/NOTIFY.FOO.D.2.C.{}.20171204172900.PDF'.format(expected_crown_text)
@freeze_time("2017-12-04 17:31:00")
def test_get_letter_pdf_filename_returns_tomorrows_filename(notify_api, mocker):
filename = get_letter_pdf_filename(reference='foo', crown=True)
assert filename == '2017-12-05/NOTIFY.FOO.D.2.C.C.20171204173100.PDF'
@pytest.mark.parametrize('personalisation', [{'name': 'test'}, None])
def test_get_letters_pdf_calls_notifications_template_preview_service_correctly(
notify_api, mocker, client, sample_letter_template, personalisation):

4
tests/app/celery/test_research_mode_tasks.py
View File

@@ -112,7 +112,7 @@ def test_failure_firetext_callback(phone_number):
def test_create_fake_letter_response_file_uploads_response_file_s3(
notify_api, mocker):
mock_s3upload = mocker.patch('app.celery.research_mode_tasks.s3upload')
filename = 'NOTIFY.20180125140000.RSP.TXT'
filename = 'NOTIFY-20180125140000-RSP.TXT'
with requests_mock.Mocker() as request_mock:
request_mock.post(
@@ -135,7 +135,7 @@ def test_create_fake_letter_response_file_uploads_response_file_s3(
def test_create_fake_letter_response_file_calls_dvla_callback_on_development(
notify_api, mocker):
mocker.patch('app.celery.research_mode_tasks.s3upload')
filename = 'NOTIFY.20180125140000.RSP.TXT'
filename = 'NOTIFY-20180125140000-RSP.TXT'
with set_config_values(notify_api, {
'NOTIFY_ENVIRONMENT': 'development'

42
tests/app/dao/notification_dao/test_notification_dao.py
View File

@@ -1769,6 +1769,48 @@ def test_dao_get_notifications_by_to_field_matches_partial_emails(sample_email_t
assert notification_2.id not in notification_ids
@pytest.mark.parametrize('search_term, expected_result_count', [
('foobar', 1),
('foo', 2),
('bar', 2),
('foo%', 1),
('%%bar', 1),
('%_', 1),
('%', 2),
('_', 1),
('/', 1),
('\\', 1),
('baz\\baz', 1),
('%foo', 0),
('%_%', 0),
('example.com', 5),
])
def test_dao_get_notifications_by_to_field_escapes(
sample_email_template,
search_term,
expected_result_count,
):
for email_address in {
'foo%_@example.com',
'%%bar@example.com',
'foobar@example.com',
'/@example.com',
'baz\\baz@example.com',
}:
create_notification(
template=sample_email_template,
to_field=email_address,
normalised_to=email_address,
)
assert len(dao_get_notifications_by_to_field(
sample_email_template.service_id,
search_term,
notification_type='email',
)) == expected_result_count
@pytest.mark.parametrize('search_term', [
'001',
'100',

20
tests/app/dao/test_api_key_dao.py
View File

@@ -1,4 +1,4 @@
from datetime import datetime
from datetime import datetime, timedelta
import pytest
from sqlalchemy.exc import IntegrityError
@@ -95,3 +95,21 @@ def test_save_api_key_should_not_create_new_service_history(sample_service):
save_model_api_key(api_key)
assert Service.get_history_model().query.count() == 1
@pytest.mark.parametrize('days_old, expected_length', [(5, 1), (8, 0)])
def test_should_not_return_revoked_api_keys_older_than_7_days(
sample_service,
days_old,
expected_length
):
expired_api_key = ApiKey(**{'service': sample_service,
'name': sample_service.name,
'created_by': sample_service.created_by,
'key_type': KEY_TYPE_NORMAL,
'expiry_date': datetime.utcnow() - timedelta(days=days_old)})
save_model_api_key(expired_api_key)
all_api_keys = get_model_api_keys(service_id=sample_service.id)
assert len(all_api_keys) == expected_length

19
tests/app/dao/test_daily_sorted_letter_dao.py
View File

@@ -20,7 +20,10 @@ def test_dao_get_daily_sorted_letter_by_billing_day(notify_db, notify_db_session
def test_dao_create_or_update_daily_sorted_letter_creates_a_new_entry(notify_db, notify_db_session):
billing_day = date(2018, 2, 1)
dsl = DailySortedLetter(billing_day=billing_day, unsorted_count=2, sorted_count=0)
dsl = DailySortedLetter(billing_day=billing_day,
file_name="Notify-201802011234.rs.txt",
unsorted_count=2,
sorted_count=0)
dao_create_or_update_daily_sorted_letter(dsl)
daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(billing_day)
@@ -35,13 +38,19 @@ def test_dao_create_or_update_daily_sorted_letter_updates_an_existing_entry(
notify_db,
notify_db_session
):
create_daily_sorted_letter(unsorted_count=2, sorted_count=3)
create_daily_sorted_letter(billing_day=date(2018, 1, 18),
file_name="Notify-20180118123.rs.txt",
unsorted_count=2,
sorted_count=3)
dsl = DailySortedLetter(billing_day=date(2018, 1, 18), unsorted_count=5, sorted_count=17)
dsl = DailySortedLetter(billing_day=date(2018, 1, 18),
file_name="Notify-20180118123.rs.txt",
unsorted_count=5,
sorted_count=17)
dao_create_or_update_daily_sorted_letter(dsl)
daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(dsl.billing_day)
assert daily_sorted_letter.unsorted_count == 7
assert daily_sorted_letter.sorted_count == 20
assert daily_sorted_letter.unsorted_count == 5
assert daily_sorted_letter.sorted_count == 17
assert daily_sorted_letter.updated_at

6
tests/app/db.py
View File

@@ -509,9 +509,13 @@ def create_invited_org_user(organisation, invited_by, email_address='invite@exam
return invited_org_user
def create_daily_sorted_letter(billing_day=date(2018, 1, 18), unsorted_count=0, sorted_count=0):
def create_daily_sorted_letter(billing_day=date(2018, 1, 18),
file_name="Notify-20180118123.rs.txt",
unsorted_count=0,
sorted_count=0):
daily_sorted_letter = DailySortedLetter(
billing_day=billing_day,
file_name=file_name,
unsorted_count=unsorted_count,
sorted_count=sorted_count
)

86
tests/app/letters/test_letter_utils.py
View File

@@ -1,6 +1,26 @@
import pytest
from datetime import datetime
from app.letters.utils import get_bucket_prefix_for_notification
import boto3
from flask import current_app
from freezegun import freeze_time
from moto import mock_s3
from app.letters.utils import get_bucket_prefix_for_notification, get_letter_pdf_filename, get_letter_pdf
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEST, PRECOMPILED_TEMPLATE_NAME
FROZEN_DATE_TIME = "2018-03-14 17:00:00"
@pytest.fixture()
@freeze_time(FROZEN_DATE_TIME)
def sample_precompiled_letter_notification_using_test_key(sample_letter_notification):
sample_letter_notification.template.hidden = True
sample_letter_notification.template.name = PRECOMPILED_TEMPLATE_NAME
sample_letter_notification.key_type = KEY_TYPE_TEST
sample_letter_notification.reference = 'foo'
sample_letter_notification.created_at = datetime.utcnow()
return sample_letter_notification
def test_get_bucket_prefix_for_notification_valid_notification(sample_notification):
@@ -13,6 +33,70 @@ def test_get_bucket_prefix_for_notification_valid_notification(sample_notificati
).upper()
@freeze_time(FROZEN_DATE_TIME)
def test_get_bucket_prefix_for_notification_precompiled_letter_using_test_key(
sample_precompiled_letter_notification_using_test_key
):
bucket_prefix = get_bucket_prefix_for_notification(
sample_precompiled_letter_notification_using_test_key, is_test_letter=True)
assert bucket_prefix == 'NOTIFY.{}'.format(
sample_precompiled_letter_notification_using_test_key.reference).upper()
def test_get_bucket_prefix_for_notification_invalid_notification():
with pytest.raises(AttributeError):
get_bucket_prefix_for_notification(None)
@pytest.mark.parametrize('crown_flag,expected_crown_text', [
(True, 'C'),
(False, 'N'),
])
@freeze_time("2017-12-04 17:29:00")
def test_get_letter_pdf_filename_returns_correct_filename(
notify_api, mocker, crown_flag, expected_crown_text):
filename = get_letter_pdf_filename(reference='foo', crown=crown_flag)
assert filename == '2017-12-04/NOTIFY.FOO.D.2.C.{}.20171204172900.PDF'.format(expected_crown_text)
@freeze_time("2017-12-04 17:29:00")
def test_get_letter_pdf_filename_returns_correct_filename_for_test_letters(
notify_api, mocker):
filename = get_letter_pdf_filename(reference='foo', crown='C', is_test_letter=True)
assert filename == 'NOTIFY.FOO.D.2.C.C.20171204172900.PDF'
@freeze_time("2017-12-04 17:31:00")
def test_get_letter_pdf_filename_returns_tomorrows_filename(notify_api, mocker):
filename = get_letter_pdf_filename(reference='foo', crown=True)
assert filename == '2017-12-05/NOTIFY.FOO.D.2.C.C.20171204173100.PDF'
@mock_s3
@pytest.mark.parametrize('bucket_config_name,filename_format', [
('TEST_LETTERS_BUCKET_NAME', 'NOTIFY.FOO.D.2.C.C.%Y%m%d%H%M%S.PDF'),
('LETTERS_PDF_BUCKET_NAME', '%Y-%m-%d/NOTIFY.FOO.D.2.C.C.%Y%m%d%H%M%S.PDF')
])
@freeze_time(FROZEN_DATE_TIME)
def test_get_letter_pdf_gets_pdf_from_correct_bucket(
sample_precompiled_letter_notification_using_test_key,
bucket_config_name,
filename_format
):
if bucket_config_name == 'LETTERS_PDF_BUCKET_NAME':
sample_precompiled_letter_notification_using_test_key.key_type = KEY_TYPE_NORMAL
bucket_name = current_app.config[bucket_config_name]
filename = datetime.utcnow().strftime(filename_format)
conn = boto3.resource('s3', region_name='eu-west-1')
conn.create_bucket(Bucket=bucket_name)
s3 = boto3.client('s3', region_name='eu-west-1')
s3.put_object(Bucket=bucket_name, Key=filename, Body=b'pdf_content')
ret = get_letter_pdf(sample_precompiled_letter_notification_using_test_key)
assert ret == b'pdf_content'

4
tests/app/notifications/rest/test_callbacks.py
View File

@@ -96,12 +96,12 @@ def test_dvla_rs_txt_file_callback_calls_update_letter_notifications_task(client
def test_dvla_rsp_txt_file_callback_calls_update_letter_notifications_task(client, mocker):
update_task = \
mocker.patch('app.notifications.notifications_letter_callback.update_letter_notifications_statuses.apply_async')
data = _sample_sns_s3_callback('NOTIFY.20170823160812.RSP.TXT')
data = _sample_sns_s3_callback('NOTIFY-20170823160812-RSP.TXT')
response = dvla_post(client, data)
assert response.status_code == 200
assert update_task.called
update_task.assert_called_with(['NOTIFY.20170823160812.RSP.TXT'], queue='notify-internal-tasks')
update_task.assert_called_with(['NOTIFY-20170823160812-RSP.TXT'], queue='notify-internal-tasks')
def test_dvla_ack_calls_does_not_call_letter_notifications_task(client, mocker):

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

@@ -15,6 +15,7 @@ from app.models import (
)
from app.dao.permissions_dao import default_service_permissions
from tests import create_authorization_header
from tests.app.db import create_service, create_organisation, create_user
def test_get_user_list(admin_request, sample_service):
@@ -36,28 +37,53 @@ def test_get_user_list(admin_request, sample_service):
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user(client, sample_service):
def test_get_user(admin_request, sample_service, sample_organisation):
"""
Tests GET endpoint '/<user_id>' to retrieve a single service.
"""
sample_user = sample_service.users[0]
header = create_authorization_header()
resp = client.get(url_for('user.get_user',
user_id=sample_user.id),
headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
sample_user.organisations = [sample_organisation]
json_resp = admin_request.get(
'user.get_user',
user_id=sample_user.id
)
expected_permissions = default_service_permissions
fetched = json_resp['data']
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert fetched['id'] == str(sample_user.id)
assert fetched['name'] == sample_user.name
assert fetched['mobile_number'] == sample_user.mobile_number
assert fetched['email_address'] == sample_user.email_address
assert fetched['state'] == sample_user.state
assert fetched['auth_type'] == SMS_AUTH_TYPE
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert fetched['permissions'].keys() == {str(sample_service.id)}
assert fetched['services'] == [str(sample_service.id)]
assert fetched['organisations'] == [str(sample_organisation.id)]
assert sorted(fetched['permissions'][str(sample_service.id)]) == sorted(expected_permissions)
def test_get_user_doesnt_return_inactive_services_and_orgs(admin_request, sample_service, sample_organisation):
"""
Tests GET endpoint '/<user_id>' to retrieve a single service.
"""
sample_service.active = False
sample_organisation.active = False
sample_user = sample_service.users[0]
sample_user.organisations = [sample_organisation]
json_resp = admin_request.get(
'user.get_user',
user_id=sample_user.id
)
fetched = json_resp['data']
assert fetched['id'] == str(sample_user.id)
assert fetched['services'] == []
assert fetched['organisations'] == []
assert fetched['permissions'] == {}
def test_post_user(client, notify_db, notify_db_session):
@@ -580,3 +606,128 @@ def test_cannot_update_user_password_using_attributes_method(admin_request, samp
_expected_status=400
)
assert resp['message']['_schema'] == ['Unknown field name password']
def test_get_orgs_and_services_nests_services(admin_request, sample_user):
org1 = create_organisation(name='org1')
org2 = create_organisation(name='org2')
service1 = create_service(service_name='service1')
service2 = create_service(service_name='service2')
service3 = create_service(service_name='service3')
org1.services = [service1, service2]
org2.services = []
sample_user.organisations = [org1, org2]
sample_user.services = [service1, service2, service3]
resp = admin_request.get('user.get_organisations_and_services_for_user', user_id=sample_user.id)
assert resp == {
'organisations': [
{
'name': org1.name,
'id': str(org1.id),
'services': [
{
'name': service1.name,
'id': str(service1.id)
},
{
'name': service2.name,
'id': str(service2.id)
}
]
},
{
'name': org2.name,
'id': str(org2.id),
'services': []
}
],
'services_without_organisations': [
{
'name': service3.name,
'id': str(service3.id)
}
]
}
def test_get_orgs_and_services_only_returns_active(admin_request, sample_user):
org1 = create_organisation(name='org1', active=True)
org2 = create_organisation(name='org2', active=False)
# in an active org
service1 = create_service(service_name='service1', active=True)
service2 = create_service(service_name='service2', active=False)
# active but in an inactive org
service3 = create_service(service_name='service3', active=True)
# not in an org
service4 = create_service(service_name='service4', active=True)
service5 = create_service(service_name='service5', active=False)
org1.services = [service1, service2]
org2.services = [service3]
sample_user.organisations = [org1, org2]
sample_user.services = [service1, service2, service3, service4, service5]
resp = admin_request.get('user.get_organisations_and_services_for_user', user_id=sample_user.id)
assert resp == {
'organisations': [
{
'name': org1.name,
'id': str(org1.id),
'services': [
{
'name': service1.name,
'id': str(service1.id)
}
]
}
],
'services_without_organisations': [
{
'name': service4.name,
'id': str(service4.id)
}
]
}
def test_get_orgs_and_services_only_shows_users_orgs_and_services(admin_request, sample_user):
other_user = create_user(email='other@user.com')
org1 = create_organisation(name='org1')
org2 = create_organisation(name='org2')
service1 = create_service(service_name='service1')
service2 = create_service(service_name='service2')
org1.services = [service1]
sample_user.organisations = [org2]
sample_user.services = [service1]
other_user.organisations = [org1, org2]
other_user.services = [service1, service2]
resp = admin_request.get('user.get_organisations_and_services_for_user', user_id=sample_user.id)
assert resp == {
'organisations': [
{
'name': org2.name,
'id': str(org2.id),
'services': []
}
],
# service1 belongs to org1, but the user doesn't know about org1
'services_without_organisations': [
{
'name': service1.name,
'id': str(service1.id)
}
]
}

33
tests/app/v2/notifications/test_post_letter_notifications.py
View File

@@ -1,4 +1,5 @@
import uuid
from unittest.mock import ANY
from flask import json
from flask import url_for
@@ -29,9 +30,13 @@ test_address = {
}
def letter_request(client, data, service_id, key_type=KEY_TYPE_NORMAL, _expected_status=201):
def letter_request(client, data, service_id, key_type=KEY_TYPE_NORMAL, _expected_status=201, precompiled=False):
if precompiled:
url = url_for('v2_notifications.post_precompiled_letter_notification')
else:
url = url_for('v2_notifications.post_notification', notification_type=LETTER_TYPE)
resp = client.post(
url_for('v2_notifications.post_notification', notification_type=LETTER_TYPE),
url,
data=json.dumps(data),
headers=[
('Content-Type', 'application/json'),
@@ -382,6 +387,30 @@ def test_post_letter_notification_is_delivered_if_in_trial_mode_and_using_test_k
assert not fake_create_letter_task.called
def test_post_letter_notification_is_delivered_and_has_pdf_uploaded_to_test_letters_bucket_using_test_key(
client,
notify_user,
mocker
):
sample_letter_service = create_service(service_permissions=['letter', 'precompiled_letter'])
s3mock = mocker.patch('app.v2.notifications.post_notifications.upload_letter_pdf')
mocker.patch('app.v2.notifications.post_notifications.pdf_page_count', return_value=1)
data = {
"reference": "letter-reference",
"content": "bGV0dGVyLWNvbnRlbnQ="
}
letter_request(
client,
data=data,
service_id=str(sample_letter_service.id),
key_type=KEY_TYPE_TEST,
precompiled=True)
notification = Notification.query.one()
assert notification.status == NOTIFICATION_DELIVERED
s3mock.assert_called_once_with(ANY, b'letter-content', is_test_letter=True)
def test_post_letter_notification_persists_notification_reply_to_text(
client, notify_db_session, mocker
):