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.
8.0 kB · 237 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238from __future__ import annotations
import jsonfrom typing import cast
import httpximport pytest
from openstatus._gen.openstatus.monitor.v1 import service_pb2 as monitor_service_pb2from openstatus._transport import _AsyncJsonTransport, _JsonTransportfrom openstatus.exceptions import ( AuthenticationError, NotFoundError, ServiceUnavailableError,)
_BASE_URL = "https://api.example.test"_DEFAULT_HEADERS = { "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "openstatus-python/test python/test", "x-openstatus-key": "test-key",}
def _build_handler( captured: list[httpx.Request], *, status: int = 200, body: str = "{}", content_type: str = "application/json",) -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response( status_code=status, content=body.encode("utf-8"), headers={"content-type": content_type}, )
return httpx.MockTransport(handler)
def test_sync_transport_targets_correct_rpc_url() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured)) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, )
assert len(captured) == 1 sent = captured[0] assert sent.method == "POST" assert str(sent.url) == f"{_BASE_URL}/rpc/openstatus.monitor.v1.MonitorService/ListMonitors"
def test_sync_transport_sends_default_headers() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured)) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, )
sent = captured[0] for header_name, header_value in _DEFAULT_HEADERS.items(): assert sent.headers[header_name] == header_value
def test_sync_transport_extra_headers_override_defaults() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured)) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, extra_headers={"x-openstatus-key": "override-key", "x-trace-id": "abc"}, )
sent = captured[0] assert sent.headers["x-openstatus-key"] == "override-key" assert sent.headers["x-trace-id"] == "abc"
def test_sync_transport_serialises_request_body_as_json() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured)) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
request_msg = monitor_service_pb2.ListMonitorHTTPResponseLogsRequest(id="abc", limit=10) jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitorHTTPResponseLogs", request_msg, monitor_service_pb2.ListMonitorHTTPResponseLogsResponse, )
body_text = captured[0].content.decode("utf-8") body = cast(dict[str, object], json.loads(body_text)) assert body == {"id": "abc", "limit": 10}
def test_sync_transport_parses_response_into_message_type() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured, body='{"totalSize":7}')) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
response = jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, )
assert isinstance(response, monitor_service_pb2.ListMonitorsResponse) assert response.total_size == 7
def test_sync_transport_raises_on_4xx_envelope() -> None: captured: list[httpx.Request] = [] transport = httpx.Client( transport=_build_handler( captured, status=401, body='{"code":"unauthenticated","message":"bad key"}', ) ) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
with pytest.raises(AuthenticationError) as excinfo: jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, ) assert excinfo.value.http_status == 401 assert str(excinfo.value) == "bad key"
def test_sync_transport_raises_service_unavailable_on_non_json_5xx() -> None: captured: list[httpx.Request] = [] transport = httpx.Client( transport=_build_handler( captured, status=502, body="<html>bad gateway</html>", content_type="text/html", ) ) jt = _JsonTransport(transport, _BASE_URL, _DEFAULT_HEADERS)
with pytest.raises(ServiceUnavailableError) as excinfo: jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, ) assert excinfo.value.raw_body == "<html>bad gateway</html>" assert excinfo.value.http_status == 502
def test_sync_transport_strips_trailing_slash_from_base_url() -> None: captured: list[httpx.Request] = [] transport = httpx.Client(transport=_build_handler(captured)) jt = _JsonTransport(transport, f"{_BASE_URL}/", _DEFAULT_HEADERS)
jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, )
assert "//rpc/" not in str(captured[0].url)
# ----- async --------------------------------------------------------------- #
async def test_async_transport_targets_correct_rpc_url() -> None: captured: list[httpx.Request] = [] async with httpx.AsyncClient(transport=_build_handler(captured)) as client: jt = _AsyncJsonTransport(client, _BASE_URL, _DEFAULT_HEADERS) await jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, ) assert str(captured[0].url) == ( f"{_BASE_URL}/rpc/openstatus.monitor.v1.MonitorService/ListMonitors" )
async def test_async_transport_raises_on_4xx_envelope() -> None: captured: list[httpx.Request] = [] async with httpx.AsyncClient( transport=_build_handler(captured, status=404, body='{"code":"not_found","message":"nope"}') ) as client: jt = _AsyncJsonTransport(client, _BASE_URL, _DEFAULT_HEADERS) with pytest.raises(NotFoundError) as excinfo: await jt.call( "openstatus.monitor.v1.MonitorService", "GetMonitor", monitor_service_pb2.GetMonitorRequest(id="x"), monitor_service_pb2.GetMonitorResponse, ) assert excinfo.value.http_status == 404
async def test_async_transport_parses_response() -> None: captured: list[httpx.Request] = [] async with httpx.AsyncClient( transport=_build_handler(captured, body='{"totalSize":3}') ) as client: jt = _AsyncJsonTransport(client, _BASE_URL, _DEFAULT_HEADERS) response = await jt.call( "openstatus.monitor.v1.MonitorService", "ListMonitors", monitor_service_pb2.ListMonitorsRequest(), monitor_service_pb2.ListMonitorsResponse, ) assert response.total_size == 3