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.
6.1 kB · 168 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169from __future__ import annotations
import httpximport pytest
from openstatus._gen.openstatus.health.v1 import health_pb2 as health_pbfrom openstatus._gen.openstatus.maintenance.v1 import service_pb2 as maintenance_pbfrom openstatus._gen.openstatus.monitor.v1 import service_pb2 as monitor_pbfrom openstatus._gen.openstatus.notification.v1 import service_pb2 as notification_pbfrom openstatus._gen.openstatus.status_page.v1 import service_pb2 as status_page_pbfrom openstatus._gen.openstatus.status_report.v1 import service_pb2 as status_report_pbfrom openstatus._transport import _AsyncJsonTransport, _JsonTransportfrom openstatus.services import ( AsyncHealthServiceClient, AsyncMaintenanceServiceClient, AsyncMonitorServiceClient, AsyncNotificationServiceClient, AsyncStatusPageServiceClient, AsyncStatusReportServiceClient, HealthServiceClient, MaintenanceServiceClient, MonitorServiceClient, NotificationServiceClient, StatusPageServiceClient, StatusReportServiceClient,)
def _capture_handler() -> tuple[list[httpx.Request], httpx.MockTransport]: captured: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, content=b"{}", headers={"content-type": "application/json"})
return captured, httpx.MockTransport(handler)
_BASE = "https://api.example.test"_HEADERS = {"Content-Type": "application/json", "Accept": "application/json"}
SYNC_CASES: list[tuple[str, str, str, object]] = [ # (service_fqn, http_method_name, expected_rpc_path_suffix, request_proto) ( "openstatus.health.v1.HealthService", "check", "Check", health_pb.CheckRequest(), ), ( "openstatus.monitor.v1.MonitorService", "list_monitors", "ListMonitors", monitor_pb.ListMonitorsRequest(), ), ( "openstatus.monitor.v1.MonitorService", "create_http_monitor", "CreateHTTPMonitor", monitor_pb.CreateHTTPMonitorRequest(), ), ( "openstatus.monitor.v1.MonitorService", "get_monitor", "GetMonitor", monitor_pb.GetMonitorRequest(id="x"), ), ( "openstatus.status_report.v1.StatusReportService", "list_status_reports", "ListStatusReports", status_report_pb.ListStatusReportsRequest(), ), ( "openstatus.status_page.v1.StatusPageService", "list_status_pages", "ListStatusPages", status_page_pb.ListStatusPagesRequest(), ), ( "openstatus.maintenance.v1.MaintenanceService", "list_maintenances", "ListMaintenances", maintenance_pb.ListMaintenancesRequest(), ), ( "openstatus.notification.v1.NotificationService", "list_notifications", "ListNotifications", notification_pb.ListNotificationsRequest(), ),]
def _resolve_sync_client(fqn: str, transport: _JsonTransport) -> object: if fqn.endswith(".HealthService"): return HealthServiceClient(transport) if fqn.endswith(".MonitorService"): return MonitorServiceClient(transport) if fqn.endswith(".StatusReportService"): return StatusReportServiceClient(transport) if fqn.endswith(".StatusPageService"): return StatusPageServiceClient(transport) if fqn.endswith(".MaintenanceService"): return MaintenanceServiceClient(transport) if fqn.endswith(".NotificationService"): return NotificationServiceClient(transport) raise AssertionError(f"unknown service fqn: {fqn}")
def _resolve_async_client(fqn: str, transport: _AsyncJsonTransport) -> object: if fqn.endswith(".HealthService"): return AsyncHealthServiceClient(transport) if fqn.endswith(".MonitorService"): return AsyncMonitorServiceClient(transport) if fqn.endswith(".StatusReportService"): return AsyncStatusReportServiceClient(transport) if fqn.endswith(".StatusPageService"): return AsyncStatusPageServiceClient(transport) if fqn.endswith(".MaintenanceService"): return AsyncMaintenanceServiceClient(transport) if fqn.endswith(".NotificationService"): return AsyncNotificationServiceClient(transport) raise AssertionError(f"unknown service fqn: {fqn}")
@pytest.mark.parametrize(("fqn", "method_name", "rpc_suffix", "request_proto"), SYNC_CASES)def test_sync_service_dispatches_to_correct_rpc( fqn: str, method_name: str, rpc_suffix: str, request_proto: object) -> None: captured, mock_transport = _capture_handler() http = httpx.Client(transport=mock_transport) jt = _JsonTransport(http, _BASE, _HEADERS) client = _resolve_sync_client(fqn, jt) method = getattr(client, method_name) method(request_proto) assert str(captured[0].url) == f"{_BASE}/rpc/{fqn}/{rpc_suffix}"
@pytest.mark.parametrize(("fqn", "method_name", "rpc_suffix", "request_proto"), SYNC_CASES)async def test_async_service_dispatches_to_correct_rpc( fqn: str, method_name: str, rpc_suffix: str, request_proto: object) -> None: captured, mock_transport = _capture_handler() async with httpx.AsyncClient(transport=mock_transport) as http: jt = _AsyncJsonTransport(http, _BASE, _HEADERS) client = _resolve_async_client(fqn, jt) method = getattr(client, method_name) await method(request_proto) assert str(captured[0].url) == f"{_BASE}/rpc/{fqn}/{rpc_suffix}"
def test_service_methods_count_matches_descriptors() -> None: # Sanity: ensure each generated client class exposes one public method per RPC. cases = [ (HealthServiceClient, 1), (MonitorServiceClient, 14), (StatusReportServiceClient, 6), (StatusPageServiceClient, 18), (MaintenanceServiceClient, 5), (NotificationServiceClient, 7), ] for cls, expected in cases: methods = [ name for name in dir(cls) if not name.startswith("_") and callable(getattr(cls, name)) ] assert len(methods) == expected, (cls.__name__, methods)