mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-20 14:29:25 -04:00
cleanup
This commit is contained in:
@@ -1,100 +0,0 @@
|
||||
from urllib import parse
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from app.cronitor import cronitor
|
||||
from tests.conftest import set_config_values
|
||||
|
||||
|
||||
def _cronitor_url(key, command):
|
||||
return parse.urlunparse(
|
||||
parse.ParseResult(
|
||||
scheme="https",
|
||||
netloc="cronitor.link",
|
||||
path="{}/{}".format(key, command),
|
||||
params="",
|
||||
query=parse.urlencode({"host": "http://localhost:6011"}),
|
||||
fragment="",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
RUN_LINK = _cronitor_url("secret", "run")
|
||||
FAIL_LINK = _cronitor_url("secret", "fail")
|
||||
COMPLETE_LINK = _cronitor_url("secret", "complete")
|
||||
|
||||
|
||||
@cronitor("hello")
|
||||
def successful_task():
|
||||
return 1
|
||||
|
||||
|
||||
@cronitor("hello")
|
||||
def crashing_task():
|
||||
raise ValueError
|
||||
|
||||
|
||||
def test_cronitor_sends_run_and_complete(notify_api, rmock):
|
||||
rmock.get(RUN_LINK, status_code=200)
|
||||
rmock.get(COMPLETE_LINK, status_code=200)
|
||||
|
||||
with set_config_values(
|
||||
notify_api, {"CRONITOR_ENABLED": True, "CRONITOR_KEYS": {"hello": "secret"}}
|
||||
):
|
||||
assert successful_task() == 1
|
||||
|
||||
assert rmock.call_count == 2
|
||||
assert rmock.request_history[0].url == RUN_LINK
|
||||
assert rmock.request_history[1].url == COMPLETE_LINK
|
||||
|
||||
|
||||
def test_cronitor_sends_run_and_fail_if_exception(notify_api, rmock):
|
||||
rmock.get(RUN_LINK, status_code=200)
|
||||
rmock.get(FAIL_LINK, status_code=200)
|
||||
|
||||
with set_config_values(
|
||||
notify_api, {"CRONITOR_ENABLED": True, "CRONITOR_KEYS": {"hello": "secret"}}
|
||||
):
|
||||
with pytest.raises(ValueError):
|
||||
crashing_task()
|
||||
|
||||
assert rmock.call_count == 2
|
||||
assert rmock.request_history[0].url == RUN_LINK
|
||||
assert rmock.request_history[1].url == FAIL_LINK
|
||||
|
||||
|
||||
def test_cronitor_does_nothing_if_cronitor_not_enabled(notify_api, rmock):
|
||||
with set_config_values(
|
||||
notify_api, {"CRONITOR_ENABLED": False, "CRONITOR_KEYS": {"hello": "secret"}}
|
||||
):
|
||||
assert successful_task() == 1
|
||||
|
||||
assert rmock.called is False
|
||||
|
||||
|
||||
def test_cronitor_does_nothing_if_name_not_recognised(notify_api, rmock, mocker):
|
||||
mock_logger = mocker.patch("app.cronitor.current_app.logger")
|
||||
|
||||
with set_config_values(
|
||||
notify_api, {"CRONITOR_ENABLED": True, "CRONITOR_KEYS": {"not-hello": "other"}}
|
||||
):
|
||||
assert successful_task() == 1
|
||||
|
||||
mock_logger.error.assert_called_with(
|
||||
"Cronitor enabled but task_name hello not found in environment"
|
||||
)
|
||||
|
||||
assert rmock.called is False
|
||||
|
||||
|
||||
def test_cronitor_doesnt_crash_if_request_fails(notify_api, rmock):
|
||||
rmock.get(RUN_LINK, exc=requests.exceptions.ConnectTimeout)
|
||||
rmock.get(COMPLETE_LINK, status_code=500)
|
||||
|
||||
with set_config_values(
|
||||
notify_api, {"CRONITOR_ENABLED": True, "CRONITOR_KEYS": {"hello": "secret"}}
|
||||
):
|
||||
assert successful_task() == 1
|
||||
|
||||
assert rmock.call_count == 2
|
||||
Reference in New Issue
Block a user