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.
7.1 kB · 199 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200from __future__ import annotations
import httpximport pytest
from openstatus import ( AsyncOpenstatusClient, ClientOptions, OpenstatusClient,)from openstatus.services import ( AsyncMonitorServiceClient, HealthServiceClient, MonitorServiceClient,)
def _silent_handler() -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, json={})
return httpx.MockTransport(handler)
def test_sync_client_exposes_namespace_path(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) monkeypatch.delenv("OPENSTATUS_API_URL", raising=False) http = httpx.Client(transport=_silent_handler()) client = OpenstatusClient(ClientOptions(http_client=http)) assert isinstance(client.health.v1.HealthService, HealthServiceClient) assert isinstance(client.monitor.v1.MonitorService, MonitorServiceClient) client.close()
def test_async_client_exposes_namespace_path(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) monkeypatch.delenv("OPENSTATUS_API_URL", raising=False) http = httpx.AsyncClient(transport=_silent_handler()) client = AsyncOpenstatusClient(ClientOptions(async_http_client=http)) assert isinstance(client.monitor.v1.MonitorService, AsyncMonitorServiceClient)
def test_env_var_fallback_for_api_key(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("OPENSTATUS_API_KEY", "env-key") captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient(ClientOptions(http_client=http)) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) assert captured[0].headers["x-openstatus-key"] == "env-key" client.close()
def test_explicit_api_key_overrides_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("OPENSTATUS_API_KEY", "env-key") captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient(ClientOptions(api_key="explicit-key", http_client=http)) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) assert captured[0].headers["x-openstatus-key"] == "explicit-key" client.close()
def test_env_var_fallback_for_base_url(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("OPENSTATUS_API_URL", "https://internal.example/api") monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient(ClientOptions(http_client=http)) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) assert str(captured[0].url).startswith("https://internal.example/api/rpc/") client.close()
def test_default_base_url_when_no_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("OPENSTATUS_API_URL", raising=False) monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient(ClientOptions(http_client=http)) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) assert str(captured[0].url).startswith("https://api.openstatus.dev/rpc/") client.close()
def test_user_supplied_http_client_is_not_closed( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) http = httpx.Client(transport=_silent_handler()) client = OpenstatusClient(ClientOptions(http_client=http)) with client: pass assert not http.is_closed
def test_sdk_owned_http_client_is_closed_on_exit( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) with OpenstatusClient() as client: assert client is not None # The owned httpx.Client should be closed; we can't introspect it directly # without accessing private attrs, so assert via behaviour: subsequent # close() should be a no-op (httpx.Client.close is idempotent). client.close()
async def test_async_sdk_owned_client_is_closed_on_aexit( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) async with AsyncOpenstatusClient() as client: assert client is not None await client.aclose()
def test_default_headers_extra_are_merged( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient( ClientOptions( http_client=http, default_headers={"x-trace-id": "abc"}, ) ) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) assert captured[0].headers["x-trace-id"] == "abc" client.close()
def test_user_agent_includes_sdk_name( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.delenv("OPENSTATUS_API_KEY", raising=False) captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json={})
http = httpx.Client(transport=httpx.MockTransport(handler)) client = OpenstatusClient(ClientOptions(http_client=http)) client.health.v1.HealthService.check( __import__( "openstatus._gen.openstatus.health.v1.health_pb2", fromlist=["CheckRequest"] ).CheckRequest() ) ua = captured[0].headers["User-Agent"] assert ua.startswith("openstatus-python/") assert " python/" in ua client.close()