darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-01 07:35:34 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into add-caching-send-to-provider

Browse Source
This commit is contained in:
Rebecca Law
2021-02-17 12:40:18 +00:00
parent dd686bd7a8 8e8601338e
commit 88036befcf
13 changed files with 660 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/app/conftest.py
View File

@@ -14,8 +14,9 @@ from app.dao.api_key_dao import save_model_api_key
from app.dao.invited_user_dao import save_invited_user
from app.dao.jobs_dao import dao_create_job
from app.dao.notifications_dao import dao_create_notification
from app.dao.organisation_dao import dao_create_organisation
from app.dao.organisation_dao import dao_create_organisation, dao_add_service_to_organisation
from app.dao.services_dao import (dao_create_service, dao_add_user_to_service)
from app.dao.broadcast_service_dao import insert_or_update_service_broadcast_settings
from app.dao.templates_dao import dao_create_template
from app.dao.users_dao import create_secret_code, create_user_code
from app.history_meta import create_history
@@ -42,7 +43,7 @@ from app.models import (
LETTER_TYPE,
SERVICE_PERMISSION_TYPES,
ServiceEmailReplyTo,
BROADCAST_TYPE
BROADCAST_TYPE,
)
from tests import create_authorization_header
from tests.app.db import (
@@ -150,7 +151,7 @@ def sample_service(notify_db_session):
@pytest.fixture(scope='function')
def sample_broadcast_service(notify_db_session):
def sample_broadcast_service(notify_db_session, broadcast_organisation):
user = create_user()
service_name = 'Sample broadcast service'
email_from = service_name.lower().replace(' ', '.')
@@ -161,12 +162,15 @@ def sample_broadcast_service(notify_db_session):
'restricted': False,
'email_from': email_from,
'created_by': user,
'crown': True
'crown': True,
'count_as_live': False,
}
service = Service.query.filter_by(name=service_name).first()
if not service:
service = Service(**data)
dao_create_service(service, user, service_permissions=[BROADCAST_TYPE])
insert_or_update_service_broadcast_settings(service, channel="severe")
dao_add_service_to_organisation(service, current_app.config['BROADCAST_ORGANISATION_ID'])
else:
if user not in service.users:
dao_add_user_to_service(service, user)
@@ -874,6 +878,16 @@ def sample_organisation(notify_db_session):
return org
@pytest.fixture
def broadcast_organisation(notify_db_session):
org = Organisation.query.get(current_app.config['BROADCAST_ORGANISATION_ID'])
if not org:
org = Organisation(id=current_app.config['BROADCAST_ORGANISATION_ID'], name='broadcast organisation')
dao_create_organisation(org)
return org
@pytest.fixture
def restore_provider_details(notify_db, notify_db_session):
"""

423
tests/app/service/test_rest.py
View File

@@ -31,6 +31,7 @@ from app.models import (
EMAIL_TYPE,
SMS_TYPE,
LETTER_TYPE,
BROADCAST_TYPE,
INTERNATIONAL_LETTERS,
INTERNATIONAL_SMS_TYPE,
INBOUND_SMS_TYPE,
@@ -243,6 +244,7 @@ def test_get_service_by_id(admin_request, sample_service):
assert json_resp['data']['email_branding'] is None
assert json_resp['data']['prefix_sms'] is True
assert json_resp['data']['allowed_broadcast_provider'] is None
assert json_resp['data']['broadcast_channel'] is None
assert set(json_resp['data'].keys()) == {
'active',
@@ -250,6 +252,7 @@ def test_get_service_by_id(admin_request, sample_service):
'billing_contact_email_addresses',
'billing_contact_names',
'billing_reference',
'broadcast_channel',
'consent_to_research',
'contact_link',
'count_as_live',
@@ -288,6 +291,38 @@ def test_get_service_by_id_returns_allowed_broadcast_provider(notify_db, admin_r
assert json_resp['data']['allowed_broadcast_provider'] == 'ee'
def test_get_service_by_id_for_broadcast_service_takes_channel_from_service_broadcast_settings(
admin_request, sample_broadcast_service
):
assert sample_broadcast_service.broadcast_channel == 'severe'
json_resp = admin_request.get('service.get_service_by_id', service_id=sample_broadcast_service.id)
assert json_resp['data']['id'] == str(sample_broadcast_service.id)
assert json_resp['data']['broadcast_channel'] == 'severe'
def test_get_service_by_id_for_service_with_broadcast_permission_sets_channel_as_test_if_no_service_broadcast_settings(
admin_request, notify_db_session
):
service = create_service(service_permissions=[BROADCAST_TYPE])
assert BROADCAST_TYPE in [p.permission for p in service.permissions]
assert service.broadcast_channel is None
json_resp = admin_request.get('service.get_service_by_id', service_id=service.id)
assert json_resp['data']['id'] == str(service.id)
assert json_resp['data']['broadcast_channel'] == 'test'
def test_get_service_by_id_for_non_broadcast_service_sets_channel_as_none(
admin_request, sample_service
):
assert BROADCAST_TYPE not in [p.permission for p in sample_service.permissions]
json_resp = admin_request.get('service.get_service_by_id', service_id=sample_service.id)
assert json_resp['data']['id'] == str(sample_service.id)
assert json_resp['data']['broadcast_channel'] is None
@pytest.mark.parametrize('detailed', [True, False])
def test_get_service_by_id_returns_organisation_type(admin_request, sample_service, detailed):
json_resp = admin_request.get('service.get_service_by_id', service_id=sample_service.id, detailed=detailed)
@@ -3645,3 +3680,391 @@ def test_get_returned_letter(admin_request, sample_letter_template):
assert not response[4]['original_file_name']
assert not response[4]['job_row_number']
assert response[4]['uploaded_letter_file_name'] == 'filename.pdf'
@pytest.mark.parametrize('channel', ["test", "severe"])
def test_set_as_broadcast_service_sets_broadcast_channel(
admin_request, sample_service, broadcast_organisation, channel
):
assert sample_service.service_broadcast_settings is None
data = {
'broadcast_channel': channel,
'service_mode': 'live',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample service'
assert result['data']['broadcast_channel'] == channel
records = ServiceBroadcastSettings.query.filter_by(service_id=sample_service.id).all()
assert len(records) == 1
assert records[0].service_id == sample_service.id
assert records[0].channel == channel
def test_set_as_broadcast_service_updates_channel_for_broadcast_service(
admin_request, sample_broadcast_service
):
assert sample_broadcast_service.broadcast_channel == "severe"
data = {
'broadcast_channel': "test",
'service_mode': 'training',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_broadcast_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample broadcast service'
assert result['data']['broadcast_channel'] == "test"
records = ServiceBroadcastSettings.query.filter_by(service_id=sample_broadcast_service.id).all()
assert len(records) == 1
assert records[0].service_id == sample_broadcast_service.id
assert records[0].channel == "test"
@pytest.mark.parametrize('channel', ["government", "extreme", "exercise", "random", ""])
def test_set_as_broadcast_service_rejects_unknown_channels(
admin_request, sample_service, broadcast_organisation, channel
):
data = {
'broadcast_channel': channel,
'service_mode': 'live',
'provider_restriction': None,
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)
def test_set_as_broadcast_service_rejects_if_no_channel(
admin_request, notify_db, sample_service, broadcast_organisation
):
data = {
'service_mode': 'training',
'provider_restriction': None,
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)
def test_set_as_broadcast_service_gives_broadcast_permission_and_removes_other_permissions(
admin_request, sample_service, broadcast_organisation
):
current_permissions = [p.permission for p in sample_service.permissions]
assert len(current_permissions) > 0
assert BROADCAST_TYPE not in current_permissions
data = {
'broadcast_channel': "severe",
'service_mode': 'training',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['permissions'] == [BROADCAST_TYPE]
permissions = ServicePermission.query.filter_by(service_id=sample_service.id).all()
assert [p.permission for p in permissions] == [BROADCAST_TYPE]
def test_set_as_broadcast_service_maintains_broadcast_permission_for_existing_broadcast_service(
admin_request, sample_broadcast_service
):
current_permissions = [p.permission for p in sample_broadcast_service.permissions]
assert current_permissions == [BROADCAST_TYPE]
data = {
'broadcast_channel': "severe",
'service_mode': 'live',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_broadcast_service.id,
_data=data,
)
assert result['data']['permissions'] == [BROADCAST_TYPE]
permissions = ServicePermission.query.filter_by(service_id=sample_broadcast_service.id).all()
assert [p.permission for p in permissions] == [BROADCAST_TYPE]
def test_set_as_broadcast_service_sets_count_as_live_to_false(
admin_request, sample_service, broadcast_organisation
):
assert sample_service.count_as_live is True
data = {
'broadcast_channel': "severe",
'service_mode': 'live',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['count_as_live'] is False
service_from_db = Service.query.filter_by(id=sample_service.id).all()[0]
assert service_from_db.count_as_live is False
def test_set_as_broadcast_service_sets_service_org_to_broadcast_org(
admin_request, sample_service, broadcast_organisation
):
assert sample_service.organisation_id != current_app.config['BROADCAST_ORGANISATION_ID']
data = {
'broadcast_channel': "severe",
'service_mode': 'training',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['organisation'] == current_app.config['BROADCAST_ORGANISATION_ID']
service_from_db = Service.query.filter_by(id=sample_service.id).all()[0]
assert str(service_from_db.organisation_id) == current_app.config['BROADCAST_ORGANISATION_ID']
def test_set_as_broadcast_service_does_not_error_if_run_on_a_service_that_is_already_a_broadcast_service(
admin_request, sample_service, broadcast_organisation
):
data = {
'broadcast_channel': "severe",
'service_mode': "live",
'provider_restriction': None,
}
for _ in range(2):
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
@freeze_time('2021-02-02')
def test_set_as_broadcast_service_sets_service_to_live_mode(
admin_request, notify_db, sample_service, broadcast_organisation
):
sample_service.restricted = True
notify_db.session.add(sample_service)
notify_db.session.commit()
assert sample_service.restricted is True
assert sample_service.go_live_at is None
data = {
'broadcast_channel': 'severe',
'service_mode': 'live',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample service'
assert result['data']['restricted'] is False
assert result['data']['go_live_at'] == '2021-02-02 00:00:00.000000'
def test_set_as_broadcast_service_doesnt_override_existing_go_live_at(
admin_request, notify_db, sample_broadcast_service
):
sample_broadcast_service.restricted = False
sample_broadcast_service.go_live_at = datetime(2021, 1, 1)
notify_db.session.add(sample_broadcast_service)
notify_db.session.commit()
assert sample_broadcast_service.restricted is False
assert sample_broadcast_service.go_live_at is not None
data = {
'broadcast_channel': 'severe',
'service_mode': 'live',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_broadcast_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample broadcast service'
assert result['data']['restricted'] is False
assert result['data']['go_live_at'] == '2021-01-01 00:00:00.000000'
def test_set_as_broadcast_service_sets_service_to_training_mode(
admin_request, notify_db, sample_broadcast_service
):
sample_broadcast_service.restricted = False
sample_broadcast_service.go_live_at = datetime(2021, 1, 1)
notify_db.session.add(sample_broadcast_service)
notify_db.session.commit()
assert sample_broadcast_service.restricted is False
assert sample_broadcast_service.go_live_at is not None
data = {
'broadcast_channel': 'severe',
'service_mode': 'training',
'provider_restriction': None,
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_broadcast_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample broadcast service'
assert result['data']['restricted'] is True
assert result['data']['go_live_at'] is None
@pytest.mark.parametrize('service_mode', ["testing", ""])
def test_set_as_broadcast_service_rejects_unknown_service_mode(
admin_request, sample_service, broadcast_organisation, service_mode
):
data = {
'broadcast_channel': 'severe',
'service_mode': service_mode,
'provider_restriction': None,
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)
def test_set_as_broadcast_service_rejects_if_no_service_mode(
admin_request, sample_service, broadcast_organisation
):
data = {
'broadcast_channel': 'severe',
'provider_restriction': None,
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)
@pytest.mark.parametrize('provider', [None, "three", "ee", "vodafone", "o2"])
def test_set_as_broadcast_service_sets_mobile_provider_restriction(
admin_request, sample_service, broadcast_organisation, provider
):
assert sample_service.service_broadcast_settings is None
data = {
'broadcast_channel': 'severe',
'service_mode': 'live',
'provider_restriction': provider
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample service'
assert result['data']['allowed_broadcast_provider'] == provider
records = ServiceBroadcastSettings.query.filter_by(service_id=sample_service.id).all()
assert len(records) == 1
assert records[0].service_id == sample_service.id
assert records[0].provider == provider
@pytest.mark.parametrize('provider', [None, "vodafone"])
def test_set_as_broadcast_service_updates_mobile_provider_restriction(
admin_request, notify_db, sample_broadcast_service, provider
):
sample_broadcast_service.service_broadcast_settings.provider = "o2"
notify_db.session.add(sample_broadcast_service)
notify_db.session.commit()
assert sample_broadcast_service.service_broadcast_settings.provider == "o2"
data = {
'broadcast_channel': 'severe',
'service_mode': 'live',
'provider_restriction': provider
}
result = admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_broadcast_service.id,
_data=data,
)
assert result['data']['name'] == 'Sample broadcast service'
assert result['data']['allowed_broadcast_provider'] == provider
records = ServiceBroadcastSettings.query.filter_by(service_id=sample_broadcast_service.id).all()
assert len(records) == 1
assert records[0].service_id == sample_broadcast_service.id
assert records[0].provider == provider
@pytest.mark.parametrize('provider', ["three, o2", "giffgaff", "", "None"])
def test_set_as_broadcast_service_rejects_unknown_provider_restriction(
admin_request, sample_service, broadcast_organisation, provider
):
data = {
'broadcast_channel': 'test',
'service_mode': 'live',
'provider_restriction': provider
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)
def test_set_as_broadcast_service_errors_if_no_mobile_provider_restriction(
admin_request, sample_service, broadcast_organisation
):
data = {
'broadcast_channel': 'severe',
'service_mode': 'live',
}
admin_request.post(
'service.set_as_broadcast_service',
service_id=sample_service.id,
_data=data,
_expected_status=400,
)

37
tests/app/v2/broadcast/sample_cap_xml_documents.py
View File

@@ -198,3 +198,40 @@ CANCEL = """
</info>
</alert>
"""
WITH_PLACEHOLDER_FOR_CONTENT = """
<alert xmlns="urn:oasis:names:tc:emergency:cap:1.2">
<identifier>50385fcb0ab7aa447bbd46d848ce8466E</identifier>
<sender>www.gov.uk/environment-agency</sender>
<sent>2020-02-16T23:01:13-00:00</sent>
<status>Actual</status>
<msgType>Alert</msgType>
<source>Flood warning service</source>
<scope>Public</scope>
<references>www.gov.uk/environment-agency,4f6d28b10ab7aa447bbd46d85f1e9effE,2020-02-16T19:20:03+00:00</references>
<info>
<language>en-GB</language>
<category>Met</category>
<event>053/055 Issue Severe Flood Warning EA</event>
<urgency>Immediate</urgency>
<severity>Severe</severity>
<certainty>Likely</certainty>
<expires>2020-02-26T23:01:14-00:00</expires>
<senderName>Environment Agency</senderName>
<description>{}</description>
<web>https://flood-warning-information.service.gov.uk</web>
<contact>0345 988 1188</contact>
<area>
<areaDesc>River Steeping in Wainfleet All Saints</areaDesc>
<polygon>53.10569,0.24453 53.10593,0.24430 53.10601,0.24375 53.10615,0.24349 53.10629,0.24356 53.10656,0.24336 53.10697,0.24354 53.10684,0.24298 53.10694,0.24264 53.10721,0.24302 53.10752,0.24310 53.10777,0.24308 53.10805,0.24320 53.10803,0.24187 53.10776,0.24085 53.10774,0.24062 53.10702,0.24056 53.10679,0.24088 53.10658,0.24071 53.10651,0.24049 53.10656,0.24022 53.10642,0.24022 53.10632,0.24052 53.10629,0.24082 53.10612,0.24093 53.10583,0.24133 53.10564,0.24178 53.10541,0.24282 53.10569,0.24453</polygon>
<geocode>
<valueName>TargetAreaCode</valueName>
<value>053FWFSTEEP4</value>
</geocode>
</area>
</info>
</alert>
"""
LONG_GSM7 = WITH_PLACEHOLDER_FOR_CONTENT.format('a' * 1396)
LONG_UCS2 = WITH_PLACEHOLDER_FOR_CONTENT.format('ŵ' * 616)

31
tests/app/v2/broadcast/test_post_broadcast.py
View File

@@ -190,3 +190,34 @@ def test_unsupported_message_types_400(
} in (
json.loads(response.get_data(as_text=True))['errors']
)
@pytest.mark.parametrize('xml_document, expected_error', (
(sample_cap_xml_documents.LONG_UCS2, (
'description must be 615 characters or fewer (because it '
'could not be GSM7 encoded)'
)),
(sample_cap_xml_documents.LONG_GSM7, (
'description must be 1,395 characters or fewer'
)),
))
def test_content_too_long_returns_400(
client,
sample_broadcast_service,
xml_document,
expected_error,
):
auth_header = create_authorization_header(service_id=sample_broadcast_service.id)
response = client.post(
path='/v2/broadcast',
data=xml_document,
headers=[('Content-Type', 'application/cap+xml'), auth_header],
)
assert json.loads(response.get_data(as_text=True)) == {
'errors': [{
'error': 'ValidationError',
'message': expected_error,
}],
'status_code': 400,
}