Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/sdk-python. Official Python SDK for openstatus
Something went wrong. Try again.
3.5 kB · 99 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100from __future__ import annotations
import httpximport pytest
from openstatus._errors import ConnectErrorMapperfrom openstatus.exceptions import ( AuthenticationError, InvalidArgumentError, NotFoundError, OpenstatusError, PermissionDeniedError, RateLimitError, ServiceUnavailableError,)
def _make_response( status: int, body: str, content_type: str = "application/json") -> httpx.Response: return httpx.Response( status_code=status, content=body.encode("utf-8"), headers={"content-type": content_type}, )
@pytest.mark.parametrize( ("code", "expected_cls"), [ ("unauthenticated", AuthenticationError), ("not_found", NotFoundError), ("invalid_argument", InvalidArgumentError), ("permission_denied", PermissionDeniedError), ("resource_exhausted", RateLimitError), ("unavailable", ServiceUnavailableError), ("internal", OpenstatusError), ("unknown_code_42", OpenstatusError), ],)def test_code_maps_to_expected_exception(code: str, expected_cls: type[OpenstatusError]) -> None: body = f'{{"code":"{code}","message":"boom"}}' response = _make_response(400, body) with pytest.raises(expected_cls) as excinfo: ConnectErrorMapper.rethrow(response) err = excinfo.value assert err.connect_code == code assert err.http_status == 400 assert err.raw_body == body assert str(err) == "boom"
def test_missing_envelope_yields_service_unavailable_with_unknown_code() -> None: response = _make_response(500, "Internal Server Error", content_type="text/plain") response.headers.update({"reason-phrase": ""}) with pytest.raises(ServiceUnavailableError) as excinfo: ConnectErrorMapper.rethrow(response) assert excinfo.value.connect_code == "unknown" assert excinfo.value.http_status == 500 assert excinfo.value.raw_body == "Internal Server Error"
def test_html_proxy_5xx_preserves_raw_body() -> None: body = "<html><body>502 Bad Gateway</body></html>" response = _make_response(502, body, content_type="text/html") with pytest.raises(ServiceUnavailableError) as excinfo: ConnectErrorMapper.rethrow(response) assert excinfo.value.raw_body == body assert excinfo.value.http_status == 502
def test_envelope_with_details_populates_details_list() -> None: body = '{"code":"invalid_argument","message":"bad","details":[{"field":"url"}]}' response = _make_response(400, body) with pytest.raises(InvalidArgumentError) as excinfo: ConnectErrorMapper.rethrow(response) assert excinfo.value.details == [{"field": "url"}]
def test_envelope_without_message_falls_back_to_reason_phrase() -> None: response = httpx.Response( status_code=404, content=b'{"code":"not_found"}', headers={"content-type": "application/json"}, ) with pytest.raises(NotFoundError) as excinfo: ConnectErrorMapper.rethrow(response) assert excinfo.value.connect_code == "not_found" assert excinfo.value.http_status == 404 assert str(excinfo.value) in {"Not Found", "HTTP 404"}
def test_envelope_with_non_dict_top_level_is_unavailable() -> None: body = '["wat"]' response = _make_response(400, body) with pytest.raises(ServiceUnavailableError) as excinfo: ConnectErrorMapper.rethrow(response) assert excinfo.value.connect_code == "unknown" assert excinfo.value.raw_body == body