mirror of
https://github.com/andrewthetechie/py-healthchecks.io.git
synced 2025-12-05 17:18:16 +01:00
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
from urllib.parse import urljoin
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Client as HTTPXClient
|
|
from httpx import Response
|
|
|
|
from healthchecks_io import CheckCreate
|
|
from healthchecks_io import CheckTrap
|
|
from healthchecks_io import CheckUpdate
|
|
from healthchecks_io import PingFailedError
|
|
from healthchecks_io import WrongClientError
|
|
|
|
|
|
@pytest.mark.respx
|
|
def test_check_trap_sync(respx_mock, test_client):
|
|
start_url = urljoin(test_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
success_url = urljoin(test_client._ping_url, "test")
|
|
respx_mock.post(success_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
|
|
with CheckTrap(test_client, uuid="test") as ct:
|
|
pass
|
|
|
|
|
|
@pytest.mark.respx
|
|
def test_check_trap_sync_failed_ping(respx_mock, test_client):
|
|
start_url = urljoin(test_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=444, text="OK"))
|
|
with pytest.raises(PingFailedError):
|
|
with CheckTrap(test_client, uuid="test") as ct:
|
|
pass
|
|
|
|
|
|
@pytest.mark.respx
|
|
def test_check_trap_sync_exception(respx_mock, test_client):
|
|
start_url = urljoin(test_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
fail_url = urljoin(test_client._ping_url, "test/fail")
|
|
respx_mock.post(fail_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
with pytest.raises(Exception):
|
|
with CheckTrap(test_client, uuid="test") as ct:
|
|
raise Exception("Exception")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.respx
|
|
async def test_check_trap_async(respx_mock, test_async_client):
|
|
start_url = urljoin(test_async_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
success_url = urljoin(test_async_client._ping_url, "test")
|
|
respx_mock.post(success_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
|
|
async with CheckTrap(test_async_client, uuid="test") as ct:
|
|
pass
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.respx
|
|
async def test_check_trap_async_failed_ping(respx_mock, test_async_client):
|
|
start_url = urljoin(test_async_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=444, text="OK"))
|
|
with pytest.raises(PingFailedError):
|
|
async with CheckTrap(test_async_client, uuid="test") as ct:
|
|
pass
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.respx
|
|
async def test_check_trap_async_exception(respx_mock, test_async_client):
|
|
start_url = urljoin(test_async_client._ping_url, "test/start")
|
|
respx_mock.post(start_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
fail_url = urljoin(test_async_client._ping_url, "test/fail")
|
|
respx_mock.post(fail_url).mock(return_value=Response(status_code=200, text="OK"))
|
|
|
|
with pytest.raises(Exception):
|
|
async with CheckTrap(test_async_client, uuid="test") as ct:
|
|
raise Exception("Exception")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_trap_wrong_client_error(test_client, test_async_client):
|
|
|
|
with pytest.raises(WrongClientError):
|
|
async with CheckTrap(test_client, uuid="test") as ct:
|
|
pass
|
|
|
|
with pytest.raises(WrongClientError):
|
|
with CheckTrap(test_async_client, uuid="test") as ct:
|
|
pass
|
|
|
|
|
|
def test_check_trap_no_uuid_or_slug(test_client):
|
|
with pytest.raises(Exception) as exc:
|
|
with CheckTrap(test_client):
|
|
pass
|
|
assert str(exc) == "Must pass a slug or an uuid"
|