mirror of
https://github.com/GSA/notifications-api.git
synced 2026-06-30 12:15:23 -04:00
7c732fa44e
We don’t store everything that comes in the CAP XML when someone creates a broadcast via the API. One thing we do store is `reference` which is a unique (to the external system) identifier for the broadcast. We show this in the front end instead of the template name, because broadcasts created from the API don’t use templates. However this ID isn’t very friendly – the Environment Agency just supply a UUID. The Environment Agency also populate the `event` field with some human readable text, for example: > 013 Issue Severe Flood Warning EA (013 is an area code which will be meaningful to the Flood Warning Service team) We should show this in the UI instead of the reference. The first step towards this is storing it in the database and returning it in the REST endpoints. Later we can have the admin app prefer reference over `identifier`, where `reference` is present. We can’t backfill this data because we don’t keep a copy of the original XML. Seems like `<event>` is a mandatory property of `<info>`, so we don’t need to worry about the field being missing (`<info> is optional in CAP but we require it because it contains stuff like the areas which we need in order to send out the broadcast`). *** https://www.pivotaltracker.com/story/show/176927060
254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
from unittest.mock import ANY
|
||
|
||
import pytest
|
||
from flask import json
|
||
|
||
from app.dao.broadcast_message_dao import (
|
||
dao_get_broadcast_message_by_id_and_service_id,
|
||
)
|
||
from tests import create_service_authorization_header
|
||
|
||
from . import sample_cap_xml_documents
|
||
|
||
|
||
def test_broadcast_for_service_without_permission_returns_400(
|
||
client,
|
||
sample_service,
|
||
):
|
||
auth_header = create_service_authorization_header(service_id=sample_service.id)
|
||
response = client.post(
|
||
path='/v2/broadcast',
|
||
data='',
|
||
headers=[('Content-Type', 'application/json'), auth_header],
|
||
)
|
||
|
||
assert response.status_code == 400
|
||
assert response.get_json()['errors'][0]['message'] == (
|
||
'Service is not allowed to send broadcast messages'
|
||
)
|
||
|
||
|
||
def test_post_broadcast_non_cap_xml_returns_415(
|
||
client,
|
||
sample_broadcast_service,
|
||
):
|
||
auth_header = create_service_authorization_header(service_id=sample_broadcast_service.id)
|
||
|
||
response = client.post(
|
||
path='/v2/broadcast',
|
||
data=json.dumps({
|
||
'content': 'This is a test',
|
||
'reference': 'abc123',
|
||
'category': 'Other',
|
||
'areas': [
|
||
{
|
||
'name': 'Hackney Marshes',
|
||
'polygons': [[
|
||
[-0.038280487060546875, 51.55738264619775],
|
||
[-0.03184318542480469, 51.553913882566754],
|
||
[-0.023174285888671875, 51.55812972989382],
|
||
[-0.023174285888671999, 51.55812972989999],
|
||
[-0.029869079589843747, 51.56165153059717],
|
||
[-0.038280487060546875, 51.55738264619775],
|
||
]],
|
||
},
|
||
],
|
||
}),
|
||
headers=[('Content-Type', 'application/json'), auth_header],
|
||
)
|
||
|
||
assert response.status_code == 415
|
||
assert json.loads(response.get_data(as_text=True)) == {
|
||
'errors': [{
|
||
'error': 'BadRequestError',
|
||
'message': 'Content type application/json not supported'
|
||
}],
|
||
'status_code': 415,
|
||
}
|
||
|
||
|
||
def test_valid_post_cap_xml_broadcast_returns_201(
|
||
client,
|
||
sample_broadcast_service,
|
||
):
|
||
auth_header = create_service_authorization_header(service_id=sample_broadcast_service.id)
|
||
|
||
response = client.post(
|
||
path='/v2/broadcast',
|
||
data=sample_cap_xml_documents.WAINFLEET,
|
||
headers=[('Content-Type', 'application/cap+xml'), auth_header],
|
||
)
|
||
assert response.status_code == 201
|
||
|
||
response_json = json.loads(response.get_data(as_text=True))
|
||
|
||
assert response_json['approved_at'] is None
|
||
assert response_json['approved_by_id'] is None
|
||
assert response_json['areas']['names'] == [
|
||
'River Steeping in Wainfleet All Saints'
|
||
]
|
||
assert response_json['cancelled_at'] is None
|
||
assert response_json['cancelled_by_id'] is None
|
||
assert response_json['content'].startswith(
|
||
'A severe flood warning has been issued. Storm Dennis'
|
||
)
|
||
assert response_json['content'].endswith(
|
||
'closely monitoring the situation throughout the night. '
|
||
)
|
||
assert response_json['reference'] == '50385fcb0ab7aa447bbd46d848ce8466E'
|
||
assert response_json['event'] == '053/055 Issue Severe Flood Warning EA'
|
||
assert response_json['created_at'] # datetime generated by the DB so can’t freeze it
|
||
assert response_json['created_by_id'] is None
|
||
assert response_json['finishes_at'] is None
|
||
assert response_json['id'] == ANY
|
||
assert response_json['personalisation'] is None
|
||
assert response_json['service_id'] == str(sample_broadcast_service.id)
|
||
|
||
assert len(response_json['areas']['simple_polygons']) == 1
|
||
assert len(response_json['areas']['simple_polygons'][0]) == 22
|
||
assert response_json['areas']['simple_polygons'][0][0] == [53.10562, 0.244127]
|
||
assert response_json['areas']['simple_polygons'][0][-1] == [53.10562, 0.244127]
|
||
assert response_json['areas']['names'] == ['River Steeping in Wainfleet All Saints']
|
||
assert 'ids' not in response_json['areas'] # only for broadcasts created in Admin
|
||
|
||
assert response_json['starts_at'] is None
|
||
assert response_json['status'] == 'pending-approval'
|
||
assert response_json['template_id'] is None
|
||
assert response_json['template_name'] is None
|
||
assert response_json['template_version'] is None
|
||
assert response_json['updated_at'] is None
|
||
|
||
|
||
@pytest.mark.parametrize("training_mode_service", [True, False])
|
||
def test_valid_post_cap_xml_broadcast_sets_stubbed_to_true_for_training_mode_services(
|
||
client,
|
||
sample_broadcast_service,
|
||
training_mode_service
|
||
):
|
||
sample_broadcast_service.restricted = training_mode_service
|
||
auth_header = create_service_authorization_header(service_id=sample_broadcast_service.id)
|
||
|
||
response = client.post(
|
||
path='/v2/broadcast',
|
||
data=sample_cap_xml_documents.WAINFLEET,
|
||
headers=[('Content-Type', 'application/cap+xml'), auth_header],
|
||
)
|
||
|
||
assert response.status_code == 201
|
||
response_json = json.loads(response.get_data(as_text=True))
|
||
|
||
broadcast_message = dao_get_broadcast_message_by_id_and_service_id(
|
||
response_json['id'], sample_broadcast_service.id
|
||
)
|
||
assert broadcast_message.stubbed == training_mode_service
|
||
|
||
|
||
@pytest.mark.parametrize('xml_document', (
|
||
'<alert>Oh no</alert>',
|
||
'<?xml version="1.0" encoding="utf-8" ?><foo><bar/></foo>',
|
||
))
|
||
def test_invalid_post_cap_xml_broadcast_returns_400(
|
||
client,
|
||
sample_broadcast_service,
|
||
xml_document,
|
||
):
|
||
auth_header = create_service_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 response.status_code == 400
|
||
assert json.loads(response.get_data(as_text=True)) == {
|
||
'errors': [{
|
||
'error': 'BadRequestError',
|
||
'message': 'Request data is not valid CAP XML'
|
||
}],
|
||
'status_code': 400,
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize('xml_document, expected_error_message', (
|
||
(sample_cap_xml_documents.CANCEL, (
|
||
'msgType Cancel is not one of [Alert]'
|
||
)),
|
||
(sample_cap_xml_documents.UPDATE, (
|
||
'msgType Update is not one of [Alert]'
|
||
)),
|
||
))
|
||
def test_unsupported_message_types_400(
|
||
client,
|
||
sample_broadcast_service,
|
||
xml_document,
|
||
expected_error_message,
|
||
):
|
||
auth_header = create_service_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 response.status_code == 400
|
||
assert {
|
||
'error': 'ValidationError',
|
||
'message': expected_error_message,
|
||
} 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_service_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,
|
||
}
|
||
|
||
|
||
def test_invalid_areas_returns_400(
|
||
client,
|
||
sample_broadcast_service
|
||
):
|
||
auth_header = create_service_authorization_header(service_id=sample_broadcast_service.id)
|
||
response = client.post(
|
||
path='/v2/broadcast',
|
||
data=sample_cap_xml_documents.MISSING_AREA_NAMES,
|
||
headers=[('Content-Type', 'application/cap+xml'), auth_header],
|
||
)
|
||
|
||
assert json.loads(response.get_data(as_text=True)) == {
|
||
'errors': [{
|
||
'error': 'ValidationError',
|
||
# the blank spaces represent the blank areaDesc in the XML
|
||
'message': 'areas does not match ([a-zA-Z1-9]+ )*[a-zA-Z1-9]+',
|
||
}],
|
||
'status_code': 400,
|
||
}
|