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

Test sanity checks by calling top-level task

Browse Source 
Since the checks are only performed in one place we can easily take
extra care to ensure this in the tests, noting that we don't need to
do any additional setup, except if no exception is raised - I've left
these tests as-is, to avoid doing more setup.

Note that we still check the happy path for when a provider message is
already sending - just in a different test [1].

[1]: 3d71815956/tests/app/celery/test_broadcast_message_tasks.py (L263)
This commit is contained in:
Ben Thorner
2021-04-19 15:58:56 +01:00
parent ee52e3e2c9
commit 936c9ebdfe
1 changed files with 20 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
tests/app/celery/test_broadcast_message_tasks.py
View File

@@ -611,7 +611,7 @@ def test_check_provider_message_should_send_doesnt_raise_if_event_hasnt_expired_
@freeze_time('2021-01-01 12:00')
def test_check_provider_message_should_send_raises_if_event_has_expired(sample_template):
def test_send_broadcast_provider_message_raises_if_event_has_expired(sample_template):
broadcast_message = create_broadcast_message(sample_template)
current_event = create_broadcast_event(
broadcast_message,
@@ -619,12 +619,12 @@ def test_check_provider_message_should_send_raises_if_event_has_expired(sample_t
transmitted_finishes_at=datetime(2021, 1, 1, 11, 59),
)
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert 'The expiry time of 2021-01-01 11:59:00 has already passed' in str(exc.value)
@freeze_time('2021-01-01 12:00')
def test_check_provider_message_should_send_raises_if_older_event_still_sending(sample_template):
def test_send_broadcast_provider_message_raises_if_older_event_still_sending(sample_template):
broadcast_message = create_broadcast_message(sample_template)
# event approved at midnight
past_succesful_event = create_broadcast_event(
@@ -650,13 +650,13 @@ def test_check_provider_message_should_send_raises_if_older_event_still_sending(
# we havent sent the previous update yet - it's still in sending - so don't try and send this one.
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert f'Previous event {past_still_sending_event.id} (type update) has not finished sending to provider ee' in str(exc.value) # noqa
@freeze_time('2021-01-01 12:00')
def test_check_provider_message_should_send_raises_if_older_event_hasnt_started_sending_yet(sample_template):
def test_send_broadcast_provider_message_raises_if_older_event_hasnt_started_sending_yet(sample_template):
broadcast_message = create_broadcast_message(sample_template)
# event approved at midnight
past_succesful_event = create_broadcast_event(
@@ -682,7 +682,7 @@ def test_check_provider_message_should_send_raises_if_older_event_hasnt_started_
# we shouldn't send the update now, because a previous event is still stuck in sending
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert f'Previous event {past_still_sending_event.id} (type update) has no provider_message for provider ee' in str(exc.value) # noqa
@@ -709,33 +709,25 @@ def test_check_provider_message_should_send_doesnt_raise_if_newer_event_not_acke
@pytest.mark.parametrize('existing_message_status', [
BroadcastProviderMessageStatus.SENDING,
pytest.param(
BroadcastProviderMessageStatus.ACK,
marks=pytest.mark.xfail(raises=BroadcastIntegrityError)
),
pytest.param(
BroadcastProviderMessageStatus.ERR,
marks=pytest.mark.xfail(raises=BroadcastIntegrityError)
),
pytest.param(
BroadcastProviderMessageStatus.TECHNICAL_FAILURE,
marks=pytest.mark.xfail(raises=BroadcastIntegrityError)
),
BroadcastProviderMessageStatus.ACK,
BroadcastProviderMessageStatus.ERR,
BroadcastProviderMessageStatus.TECHNICAL_FAILURE,
])
def test_check_provider_message_should_send_raises_if_current_event_already_has_provider_message_not_in_sending(
def test_send_broadcast_provider_message_raises_if_current_event_already_has_provider_message_not_in_sending(
sample_template,
existing_message_status
):
broadcast_message = create_broadcast_message(sample_template)
current_event = create_broadcast_event(broadcast_message, message_type='alert')
create_broadcast_provider_message(current_event, provider='ee', status=existing_message_status)
check_provider_message_should_send(current_event, 'ee')
with pytest.raises(BroadcastIntegrityError) as exc:
send_broadcast_provider_message(current_event.id, 'ee')
assert f'in status {existing_message_status}' in str(exc.value)
def test_check_provider_message_should_send_raises_if_service_is_suspended(
def test_send_broadcast_provider_message_raises_if_service_is_suspended(
sample_broadcast_service,
):
sample_broadcast_service.active = False
@@ -743,12 +735,12 @@ def test_check_provider_message_should_send_raises_if_service_is_suspended(
current_event = create_broadcast_event(broadcast_message, message_type='alert')
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert 'service is suspended' in str(exc.value)
def test_check_provider_message_should_send_raises_if_service_is_not_live(
def test_send_broadcast_provider_message_raises_if_service_is_not_live(
sample_broadcast_service,
):
sample_broadcast_service.restricted = True
@@ -756,19 +748,19 @@ def test_check_provider_message_should_send_raises_if_service_is_not_live(
current_event = create_broadcast_event(broadcast_message, message_type='alert')
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert 'service is not live' in str(exc.value)
def test_check_provider_message_should_send_raises_if_message_is_stubbed(
def test_send_broadcast_provider_message_raises_if_message_is_stubbed(
sample_template,
):
broadcast_message = create_broadcast_message(sample_template, stubbed=True)
current_event = create_broadcast_event(broadcast_message, message_type='alert')
with pytest.raises(BroadcastIntegrityError) as exc:
check_provider_message_should_send(current_event, 'ee')
send_broadcast_provider_message(current_event.id, 'ee')
assert 'message is stubbed' in str(exc.value)