darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-23 17:01:35 -05:00
Code Issues Packages Projects Releases Wiki Activity

Move validate_and_update_broadcast_message_status to a utils file

Browse Source 
This is because that function is used both when broadcast status
is updated via API and via admin, so it's a shared resource.

Also move and update tests for updating broadcast message status
so things are tested at source and repetition is avoided.
This commit is contained in:
Pea Tyczynska
2022-01-19 16:21:09 +00:00
parent c9afb2f038
commit 52dbdb7518
6 changed files with 420 additions and 409 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
tests/app/broadcast_message/test_rest.py
View File

@@ -620,91 +620,6 @@ def test_update_broadcast_message_allows_service_user_and_platform_admin_to_canc
assert cancel_event.transmitted_content == {"body": "emergency broadcast"}
def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_queues_task(
admin_request,
sample_broadcast_service,
mocker
):
t = create_template(sample_broadcast_service, BROADCAST_TYPE, content='emergency broadcast')
bm = create_broadcast_message(
t,
status=BroadcastStatusType.PENDING_APPROVAL,
areas={
"ids": ["london"],
"simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]
}
)
approver = create_user(email='approver@gov.uk')
sample_broadcast_service.users.append(approver)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
service_id=t.service_id,
broadcast_message_id=bm.id,
_expected_status=200
)
assert response['status'] == BroadcastStatusType.BROADCASTING
assert response['approved_at'] is not None
assert response['approved_by_id'] == str(approver.id)
assert len(bm.events) == 1
alert_event = bm.events[0]
mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='broadcast-tasks')
assert alert_event.service_id == sample_broadcast_service.id
assert alert_event.transmitted_areas == bm.areas
assert alert_event.message_type == BroadcastEventMessageType.ALERT
assert alert_event.transmitted_finishes_at == bm.finishes_at
assert alert_event.transmitted_content == {"body": "emergency broadcast"}
@pytest.mark.parametrize('broadcast_message_stubbed, service_restricted_before_approval', [
(True, True),
(True, False),
(False, True),
])
def test_update_broadcast_message_status_updates_details_but_does_not_queue_task_if_bm_is_stubbed_or_service_not_live(
admin_request,
sample_broadcast_service,
mocker,
broadcast_message_stubbed,
service_restricted_before_approval,
):
sample_broadcast_service.restricted = broadcast_message_stubbed
t = create_template(sample_broadcast_service, BROADCAST_TYPE, content='emergency broadcast')
bm = create_broadcast_message(
t,
status=BroadcastStatusType.PENDING_APPROVAL,
areas={"ids": ["london"], "simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]},
stubbed=broadcast_message_stubbed
)
approver = create_user(email='approver@gov.uk')
sample_broadcast_service.users.append(approver)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
sample_broadcast_service.restricted = service_restricted_before_approval
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
service_id=t.service_id,
broadcast_message_id=bm.id,
_expected_status=200
)
assert response['status'] == BroadcastStatusType.BROADCASTING
assert response['approved_at'] is not None
assert response['approved_by_id'] == str(approver.id)
# The broadcast can be approved, but does not create a broadcast_event in the database or put a task on the queue
assert len(bm.events) == 0
assert len(mock_task.mock_calls) == 0
def test_update_broadcast_message_status_aborts_if_service_is_suspended(
admin_request,
sample_broadcast_service,
@@ -721,123 +636,6 @@ def test_update_broadcast_message_status_aborts_if_service_is_suspended(
)
def test_update_broadcast_message_status_creates_event_with_correct_content_if_broadcast_has_no_template(
admin_request,
sample_broadcast_service,
mocker
):
bm = create_broadcast_message(
service=sample_broadcast_service,
template=None,
content='tailor made emergency broadcast content',
status=BroadcastStatusType.PENDING_APPROVAL,
areas={
"ids": ["london"],
"simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]
}
)
approver = create_user(email='approver@gov.uk')
sample_broadcast_service.users.append(approver)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
service_id=sample_broadcast_service.id,
broadcast_message_id=bm.id,
_expected_status=200
)
assert response['status'] == BroadcastStatusType.BROADCASTING
assert len(bm.events) == 1
alert_event = bm.events[0]
mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='broadcast-tasks')
assert alert_event.transmitted_content == {"body": "tailor made emergency broadcast content"}
@pytest.mark.parametrize('is_platform_admin', [True, False])
def test_update_broadcast_message_status_rejects_approval_from_creator(
admin_request,
sample_broadcast_service,
mocker,
is_platform_admin
):
t = create_template(sample_broadcast_service, BROADCAST_TYPE)
bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
user = sample_broadcast_service.created_by
user.platform_admin = is_platform_admin
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(t.created_by_id)},
service_id=t.service_id,
broadcast_message_id=bm.id,
_expected_status=400
)
assert mock_task.called is False
assert 'cannot approve their own broadcast' in response['message']
def test_update_broadcast_message_status_rejects_approval_of_broadcast_with_no_areas(
admin_request,
sample_broadcast_service,
mocker
):
template = create_template(sample_broadcast_service, BROADCAST_TYPE)
broadcast = create_broadcast_message(template, status=BroadcastStatusType.PENDING_APPROVAL)
approver = create_user(email='approver@gov.uk')
sample_broadcast_service.users.append(approver)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
service_id=template.service_id,
broadcast_message_id=broadcast.id,
_expected_status=400
)
assert mock_task.called is False
assert response[
'message'
] == f'broadcast_message {broadcast.id} has no selected areas and so cannot be broadcasted.'
def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_own_message(
notify_db,
admin_request,
sample_broadcast_service,
mocker
):
sample_broadcast_service.restricted = True
t = create_template(sample_broadcast_service, BROADCAST_TYPE)
bm = create_broadcast_message(
t,
status=BroadcastStatusType.PENDING_APPROVAL,
areas={"ids": ["london"], "simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]}
)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(t.created_by_id)},
service_id=t.service_id,
broadcast_message_id=bm.id,
_expected_status=200
)
assert response['status'] == BroadcastStatusType.BROADCASTING
assert response['approved_at'] is not None
assert response['created_by_id'] == str(t.created_by_id)
assert response['approved_by_id'] == str(t.created_by_id)
assert not mock_task.called
def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_service(
admin_request,
sample_broadcast_service,
@@ -858,56 +656,3 @@ def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_
assert mock_task.called is False
assert 'cannot update broadcast' in response['message']
@pytest.mark.parametrize('current_status, new_status', [
(BroadcastStatusType.DRAFT, BroadcastStatusType.DRAFT),
(BroadcastStatusType.DRAFT, BroadcastStatusType.BROADCASTING),
(BroadcastStatusType.DRAFT, BroadcastStatusType.CANCELLED),
(BroadcastStatusType.PENDING_APPROVAL, BroadcastStatusType.PENDING_APPROVAL),
(BroadcastStatusType.PENDING_APPROVAL, BroadcastStatusType.CANCELLED),
(BroadcastStatusType.PENDING_APPROVAL, BroadcastStatusType.COMPLETED),
(BroadcastStatusType.REJECTED, BroadcastStatusType.REJECTED),
(BroadcastStatusType.REJECTED, BroadcastStatusType.BROADCASTING),
(BroadcastStatusType.REJECTED, BroadcastStatusType.CANCELLED),
(BroadcastStatusType.REJECTED, BroadcastStatusType.COMPLETED),
(BroadcastStatusType.BROADCASTING, BroadcastStatusType.DRAFT),
(BroadcastStatusType.BROADCASTING, BroadcastStatusType.PENDING_APPROVAL),
(BroadcastStatusType.BROADCASTING, BroadcastStatusType.BROADCASTING),
(BroadcastStatusType.COMPLETED, BroadcastStatusType.DRAFT),
(BroadcastStatusType.COMPLETED, BroadcastStatusType.PENDING_APPROVAL),
(BroadcastStatusType.COMPLETED, BroadcastStatusType.BROADCASTING),
(BroadcastStatusType.COMPLETED, BroadcastStatusType.CANCELLED),
(BroadcastStatusType.CANCELLED, BroadcastStatusType.DRAFT),
(BroadcastStatusType.CANCELLED, BroadcastStatusType.PENDING_APPROVAL),
(BroadcastStatusType.CANCELLED, BroadcastStatusType.BROADCASTING),
(BroadcastStatusType.CANCELLED, BroadcastStatusType.COMPLETED),
])
def test_update_broadcast_message_status_restricts_status_transitions_to_explicit_list(
admin_request,
sample_broadcast_service,
mocker,
current_status,
new_status
):
t = create_template(sample_broadcast_service, BROADCAST_TYPE)
bm = create_broadcast_message(t, status=current_status)
approver = create_user(email='approver@gov.uk')
sample_broadcast_service.users.append(approver)
mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
response = admin_request.post(
'broadcast_message.update_broadcast_message_status',
_data={'status': new_status, 'created_by': str(approver.id)},
service_id=t.service_id,
broadcast_message_id=bm.id,
_expected_status=400
)
assert mock_task.called is False
assert f'from {current_status} to {new_status}' in response['message']