diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/adapter.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/adapter.py index a08091e8f4..da6cf6092c 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/adapter.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/adapter.py @@ -43,5 +43,6 @@ def client_from_platform( return client_cls( base_url=str(platform.base_url).rstrip("/"), workspace=platform.workspace, + default_headers=platform._custom_headers, # type: ignore[arg-type] http_client=platform._client, # type: ignore[arg-type] ) diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/client.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/client.py index f337dd200a..a2d34d403f 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/client.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/client.py @@ -18,6 +18,7 @@ import asyncio import inspect +import json import time from collections.abc import Mapping from pathlib import Path @@ -28,6 +29,7 @@ StaticToken, TokenProvider, ) +from nemo_platform_plugin.client.errors import raise_for_status from nemo_platform_plugin.client.response import ( AsyncNemoBinaryResponse, AsyncNemoPaginatedResponse, @@ -115,11 +117,13 @@ def __init__( workspace: str | None = None, auth: TokenProvider | str | None = None, retry: RetryPolicy | None = None, + default_headers: Mapping[str, str] | None = None, ) -> None: self._base_url = base_url.rstrip("/") self._workspace = workspace self._auth: TokenProvider | None = StaticToken(auth) if isinstance(auth, str) else auth self._retry = retry + self._default_headers = dict(default_headers) if default_headers else {} @property def base_url(self) -> str: @@ -158,6 +162,8 @@ def _resolve_path(self, request: PreparedRequest) -> str: def _request_headers(self, request: PreparedRequest) -> dict[str, str] | None: headers: dict[str, str] = {} + if self._default_headers: + headers.update(self._default_headers) if request.content_type is not None: headers["Content-Type"] = request.content_type if request.extra_headers: @@ -174,33 +180,19 @@ def _is_paginated(self, request: PreparedRequest) -> bool: return get_origin(request.response_type) is Paginated def _resolve_query_params(self, request: PreparedRequest) -> dict[str, str | int | bool] | None: - """Filter out None values from query params for httpx.""" + """Filter out None values and JSON-serialize dicts/lists in query params.""" if request.query_params is None: return None - filtered = {k: v for k, v in request.query_params.items() if v is not None} + filtered = {} + for k, v in request.query_params.items(): + if v is None: + continue + if isinstance(v, (dict, list)): + filtered[k] = json.dumps(v) + else: + filtered[k] = v return filtered or None - def _apply_client_options(self, request: PreparedRequest, response: NemoResponse) -> NemoResponse: - """Apply blessed client options (e.g. ``exist_ok``) to the response. - - Options are stashed on ``PreparedRequest.client_options`` by the - endpoint decorator and applied here after the HTTP call completes. - """ - if not request.client_options: - return response - - if request.client_options.get("exist_ok"): - if response.http_response.status_code == 409: - body = response.body - if body is None and request.response_type is not None: - try: - body = request.response_type.model_validate(response.http_response.json()) - except (ValueError, TypeError): - pass - return NemoResponse(http_response=response.http_response, body=body, request=request) - - return response - class NemoClient(BaseNemoClient): """Sync HTTP client for NeMo Platform APIs.""" @@ -216,7 +208,9 @@ def __init__( retry: RetryPolicy | None = None, http_client: httpx.Client | None = None, ) -> None: - super().__init__(base_url=base_url, workspace=workspace, auth=auth, retry=retry) + super().__init__( + base_url=base_url, workspace=workspace, auth=auth, retry=retry, default_headers=default_headers + ) self._http = http_client or httpx.Client( headers=dict(default_headers) if default_headers else None, timeout=timeout, @@ -339,11 +333,14 @@ def send( ) raw = self._request_with_retry(request, url, req_headers, params, resolved_retry) + # NOTE: client_options (e.g. exist_ok) from PreparedRequest are not + # acted on here yet — see AIRCORE-866 for the planned server-side + # fix that would let the client handle them properly. + raise_for_status(raw) body = None - if raw.is_success and request.response_type is not None: + if request.response_type is not None: body = request.response_type.model_validate(raw.json()) - response = NemoResponse(http_response=raw, body=body, request=request) - return self._apply_client_options(request, response) + return NemoResponse(http_response=raw, body=body, request=request) def _request_with_retry( self, @@ -408,7 +405,9 @@ def __init__( retry: RetryPolicy | None = None, http_client: httpx.AsyncClient | None = None, ) -> None: - super().__init__(base_url=base_url, workspace=workspace, auth=auth, retry=retry) + super().__init__( + base_url=base_url, workspace=workspace, auth=auth, retry=retry, default_headers=default_headers + ) self._http = http_client or httpx.AsyncClient( headers=dict(default_headers) if default_headers else None, timeout=timeout, @@ -523,11 +522,14 @@ async def send( ) raw = await self._request_with_retry(request, url, req_headers, params, resolved_retry) + # NOTE: client_options (e.g. exist_ok) from PreparedRequest are not + # acted on here yet — see AIRCORE-866 for the planned server-side + # fix that would let the client handle them properly. + raise_for_status(raw) body = None - if raw.is_success and request.response_type is not None: + if request.response_type is not None: body = request.response_type.model_validate(raw.json()) - response = NemoResponse(http_response=raw, body=body, request=request) - return self._apply_client_options(request, response) + return NemoResponse(http_response=raw, body=body, request=request) async def _request_with_retry( self, diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/endpoint.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/endpoint.py index aeb4483dfb..0ca1f8aff2 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/endpoint.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/endpoint.py @@ -129,7 +129,7 @@ def _build_prepared_request( elif name == "body": if not isinstance(value, BaseModel): raise TypeError(f"body must be a BaseModel instance, got {type(value).__name__}") - content = value.model_dump_json().encode() + content = value.model_dump_json(exclude_unset=True).encode() content_type = "application/json" elif name == "content": content = value diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/errors.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/errors.py new file mode 100644 index 0000000000..e8d67b1c9f --- /dev/null +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/errors.py @@ -0,0 +1,112 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""HTTP error hierarchy for the NemoClient. + +Provides :class:`NemoHTTPError` and status-code-specific subclasses +(e.g. :class:`NotFoundError`, :class:`ConflictError`) raised by +:func:`raise_for_status` on non-2xx responses. +""" + +from __future__ import annotations + +import httpx + + +class NemoHTTPError(Exception): + """Raised on non-2xx HTTP responses. + + Attributes: + http_response: The raw httpx response. + status_code: The HTTP status code. + detail: A human-readable error message extracted from the response + body (``{"detail": "..."}`` convention used by FastAPI / NeMo + Platform), or the raw response text as a fallback. + body: The parsed JSON response body, or None. + """ + + def __init__(self, http_response: httpx.Response) -> None: + self.http_response = http_response + self.status_code = http_response.status_code + self.detail = self._extract_detail(http_response) + self.body = self._extract_body(http_response) + super().__init__(f"HTTP {self.status_code}: {self.detail}") + + @staticmethod + def _extract_body(resp: httpx.Response) -> object | None: + try: + return resp.json() + except Exception: + return None + + @staticmethod + def _extract_detail(resp: httpx.Response) -> str: + try: + body = resp.json() + if isinstance(body, dict) and isinstance(body.get("detail"), str): + return body["detail"] + except Exception: + pass + try: + return resp.text + except Exception: + return resp.reason_phrase or f"HTTP {resp.status_code}" + + +# --------------------------------------------------------------------------- +# Status-code-specific errors +# --------------------------------------------------------------------------- + + +class BadRequestError(NemoHTTPError): + """HTTP 400""" + + +class AuthenticationError(NemoHTTPError): + """HTTP 401""" + + +class PermissionDeniedError(NemoHTTPError): + """HTTP 403""" + + +class NotFoundError(NemoHTTPError): + """HTTP 404""" + + +class ConflictError(NemoHTTPError): + """HTTP 409""" + + +class UnprocessableEntityError(NemoHTTPError): + """HTTP 422""" + + +class RateLimitError(NemoHTTPError): + """HTTP 429""" + + +class InternalServerError(NemoHTTPError): + """HTTP 500+""" + + +_STATUS_CODE_TO_ERROR: dict[int, type[NemoHTTPError]] = { + 400: BadRequestError, + 401: AuthenticationError, + 403: PermissionDeniedError, + 404: NotFoundError, + 409: ConflictError, + 422: UnprocessableEntityError, + 429: RateLimitError, + 500: InternalServerError, +} + + +def raise_for_status(http_response: httpx.Response) -> None: + """Raise status-code-specific NemoHTTPError subclass for non-2xx responses.""" + if 200 <= http_response.status_code < 300: + return + error_cls = _STATUS_CODE_TO_ERROR.get(http_response.status_code, NemoHTTPError) + if error_cls is NemoHTTPError and http_response.status_code >= 500: + error_cls = InternalServerError + raise error_cls(http_response) diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/response.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/response.py index 92f7971bc1..7b0d89e048 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/response.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/response.py @@ -6,15 +6,32 @@ from __future__ import annotations from collections.abc import AsyncIterator, Callable, Coroutine, Iterator -from contextlib import AbstractAsyncContextManager, AbstractContextManager +from contextlib import AbstractAsyncContextManager, AbstractContextManager, asynccontextmanager, contextmanager from dataclasses import dataclass -from types import TracebackType from typing import Any, Generic, TypeVar import httpx +from nemo_platform_plugin.client.errors import raise_for_status from nemo_platform_plugin.client.types import OffsetPagination, PaginationStrategy, PreparedRequest from pydantic import BaseModel + +def _parse_stream_line(line: str, headers: httpx.Headers) -> str | None: + """Extract a JSON payload from a stream line, or ``None`` to skip. + + Handles both NDJSON (pass-through) and SSE framing (strips ``data:`` + prefix, skips non-data fields like ``event:``, ``id:``, comments). + """ + line = line.strip() + if not line: + return None + if "text/event-stream" in headers.get("content-type", ""): + if line.startswith("data:"): + return line[5:].strip() + return None + return line + + ResponseT = TypeVar("ResponseT") ModelT = TypeVar("ModelT", bound=BaseModel) @@ -37,9 +54,11 @@ class NemoResponse(Generic[ResponseT]): request: PreparedRequest def data(self) -> ResponseT: - """Return the body if the status is 2xx, otherwise raise.""" - if not (200 <= self.http_response.status_code < 300): - raise NemoHTTPError(self.http_response) + """Return the parsed response body. + + Since ``send()`` raises on non-2xx, this is a convenience accessor + equivalent to ``.body``. + """ return self.body @@ -51,48 +70,49 @@ def data(self) -> ResponseT: class NemoBinaryResponse: """Sync response for binary download endpoints. - Use as a context manager:: + For simple reads:: - with client.send(endpoints.download(...)) as resp: - data = resp.read() # all bytes at once - # or: for chunk in resp # iterate chunks + resp = client.send(endpoints.download(...)) + data = resp.read() + + For streaming chunks:: + + with resp.stream() as chunks: + for chunk in chunks: + f.write(chunk) """ def __init__(self, stream_ctx: AbstractContextManager[httpx.Response], request: PreparedRequest) -> None: self._stream_ctx = stream_ctx - self._response: httpx.Response | None = None self.request = request - @property - def http_response(self) -> httpx.Response: - assert self._response is not None, "Must enter context manager before accessing response" - return self._response - def read(self) -> bytes: """Read and return the entire response body as bytes.""" - return self.http_response.read() + with self._stream_ctx as raw: + data = raw.read() + raise_for_status(raw) + return data - def __iter__(self) -> Iterator[bytes]: - return self.http_response.iter_bytes() - - def __enter__(self) -> NemoBinaryResponse: - self._response = self._stream_ctx.__enter__() - self._response.raise_for_status() - return self - - def __exit__( - self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None - ) -> None: - self._stream_ctx.__exit__(exc_type, exc_val, exc_tb) + @contextmanager + def stream(self) -> Iterator[Iterator[bytes]]: + """Yield an iterator of byte chunks.""" + with self._stream_ctx as raw: + raise_for_status(raw) + yield raw.iter_bytes() class NemoStreamResponse(Generic[ModelT]): """Sync response for SSE/NDJSON streaming endpoints. - Use as a context manager:: + Handles both NDJSON (``application/x-ndjson``) and SSE + (``text/event-stream``) framing automatically based on the + response ``Content-Type``. SSE ``data:`` prefixes are stripped + before JSON parsing. - with client.send(ChatEndpoint(...)) as resp: - for chunk in resp: + Use via :meth:`stream`:: + + with client.send(ChatEndpoint(...)).stream() as chunks: + for chunk in chunks: print(chunk.text) """ @@ -104,29 +124,21 @@ def __init__( ) -> None: self._stream_ctx = stream_ctx self._model_type = model_type - self._response: httpx.Response | None = None self.request = request - @property - def http_response(self) -> httpx.Response: - assert self._response is not None, "Must enter context manager before accessing response" - return self._response + @contextmanager + def stream(self) -> Iterator[Iterator[ModelT]]: + """Yield an iterator of parsed model objects.""" + with self._stream_ctx as raw: + raise_for_status(raw) - def __iter__(self) -> Iterator[ModelT]: - for line in self.http_response.iter_lines(): - line = line.strip() - if line: - yield self._model_type.model_validate_json(line) - - def __enter__(self) -> NemoStreamResponse[ModelT]: - self._response = self._stream_ctx.__enter__() - self._response.raise_for_status() - return self - - def __exit__( - self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None - ) -> None: - self._stream_ctx.__exit__(exc_type, exc_val, exc_tb) + def _iter() -> Iterator[ModelT]: + for line in raw.iter_lines(): + payload = _parse_stream_line(line, raw.headers) + if payload is not None: + yield self._model_type.model_validate_json(payload) + + yield _iter() # --------------------------------------------------------------------------- @@ -137,49 +149,47 @@ def __exit__( class AsyncNemoBinaryResponse: """Async response for binary download endpoints. - Use as an async context manager:: + For simple reads:: + + resp = await client.send(endpoints.download(...)) + data = await resp.read() + + For streaming chunks:: - async with client.send(endpoints.download(...)) as resp: - data = await resp.read() # all bytes at once - # or: async for chunk in resp # iterate chunks + async with resp.stream() as chunks: + async for chunk in chunks: + f.write(chunk) """ def __init__(self, stream_ctx: AbstractAsyncContextManager[httpx.Response], request: PreparedRequest) -> None: self._stream_ctx = stream_ctx - self._response: httpx.Response | None = None self.request = request - @property - def http_response(self) -> httpx.Response: - assert self._response is not None, "Must enter async context manager before accessing response" - return self._response - async def read(self) -> bytes: """Read and return the entire response body as bytes.""" - return await self.http_response.aread() + async with self._stream_ctx as raw: + data = await raw.aread() + raise_for_status(raw) + return data - async def __aiter__(self) -> AsyncIterator[bytes]: - async for chunk in self.http_response.aiter_bytes(): - yield chunk - - async def __aenter__(self) -> AsyncNemoBinaryResponse: - self._response = await self._stream_ctx.__aenter__() - self._response.raise_for_status() - return self - - async def __aexit__( - self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None - ) -> None: - await self._stream_ctx.__aexit__(exc_type, exc_val, exc_tb) + @asynccontextmanager + async def stream(self) -> AsyncIterator[AsyncIterator[bytes]]: + """Yield an async iterator of byte chunks.""" + async with self._stream_ctx as raw: + raise_for_status(raw) + yield raw.aiter_bytes() class AsyncNemoStreamResponse(Generic[ModelT]): """Async response for SSE/NDJSON streaming endpoints. - Use as an async context manager:: + Handles both NDJSON and SSE framing automatically based on the + response ``Content-Type``. See :class:`NemoStreamResponse` for details. - async with client.send(ChatEndpoint(...)) as resp: - async for chunk in resp: + Use via :meth:`stream`:: + + async with (await client.send(ChatEndpoint(...))).stream() as chunks: + async for chunk in chunks: print(chunk.text) """ @@ -191,29 +201,21 @@ def __init__( ) -> None: self._stream_ctx = stream_ctx self._model_type = model_type - self._response: httpx.Response | None = None self.request = request - @property - def http_response(self) -> httpx.Response: - assert self._response is not None, "Must enter async context manager before accessing response" - return self._response + @asynccontextmanager + async def stream(self) -> AsyncIterator[AsyncIterator[ModelT]]: + """Yield an async iterator of parsed model objects.""" + async with self._stream_ctx as raw: + raise_for_status(raw) - async def __aiter__(self) -> AsyncIterator[ModelT]: - async for line in self.http_response.aiter_lines(): - line = line.strip() - if line: - yield self._model_type.model_validate_json(line) - - async def __aenter__(self) -> AsyncNemoStreamResponse[ModelT]: - self._response = await self._stream_ctx.__aenter__() - self._response.raise_for_status() - return self - - async def __aexit__( - self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None - ) -> None: - await self._stream_ctx.__aexit__(exc_type, exc_val, exc_tb) + async def _iter() -> AsyncIterator[ModelT]: + async for line in raw.aiter_lines(): + payload = _parse_stream_line(line, raw.headers) + if payload is not None: + yield self._model_type.model_validate_json(payload) + + yield _iter() # --------------------------------------------------------------------------- @@ -284,7 +286,7 @@ def http_response(self) -> httpx.Response: def _parse_page(self, raw: httpx.Response) -> tuple[list[ModelT], dict]: """Parse a page response into (items, raw_body).""" - raw.raise_for_status() + raise_for_status(raw) body = raw.json() items = [self._model_type.model_validate(item) for item in self._strategy.extract_items(body)] return items, body @@ -336,7 +338,7 @@ def http_response(self) -> httpx.Response: def _parse_page(self, raw: httpx.Response) -> tuple[list[ModelT], dict]: """Parse a page response into (items, raw_body).""" - raw.raise_for_status() + raise_for_status(raw) body = raw.json() items = [self._model_type.model_validate(item) for item in self._strategy.extract_items(body)] return items, body @@ -360,36 +362,3 @@ async def __aiter__(self) -> AsyncIterator[ModelT]: yield item current = next_page next_page = self._strategy.next_page(body, current) - - -# --------------------------------------------------------------------------- -# Errors -# --------------------------------------------------------------------------- - - -class NemoHTTPError(Exception): - """Raised by :meth:`NemoResponse.data` on non-2xx responses. - - Attributes: - http_response: The raw httpx response. - status_code: The HTTP status code. - detail: A human-readable error message extracted from the response - body (``{"detail": "..."}`` convention used by FastAPI / NeMo - Platform), or the raw response text as a fallback. - """ - - def __init__(self, http_response: httpx.Response) -> None: - self.http_response = http_response - self.status_code = http_response.status_code - self.detail = self._extract_detail(http_response) - super().__init__(f"HTTP {self.status_code}: {self.detail}") - - @staticmethod - def _extract_detail(resp: httpx.Response) -> str: - try: - body = resp.json() - if isinstance(body, dict) and isinstance(body.get("detail"), str): - return body["detail"] - except Exception: - pass - return resp.text diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/types.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/types.py index 1e2a6c72b0..d1eb63df96 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/types.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/client/types.py @@ -167,6 +167,7 @@ def list_widgets(...) -> Paginated[Widget, MyPagination]: ... # Unknown parameters in an endpoint signature trigger a ``TypeError`` # at decoration time. BLESSED_CLIENT_PARAMS: dict[str, type] = { + # Declared but not yet acted on by the client — see AIRCORE-866. "exist_ok": bool, } diff --git a/packages/nemo_platform_plugin/tests/client/test_client.py b/packages/nemo_platform_plugin/tests/client/test_client.py index 2ea7569ce6..d415b6e458 100644 --- a/packages/nemo_platform_plugin/tests/client/test_client.py +++ b/packages/nemo_platform_plugin/tests/client/test_client.py @@ -9,7 +9,8 @@ import pytest from nemo_platform_plugin.client.client import AsyncNemoClient, NemoClient from nemo_platform_plugin.client.endpoint import delete, get, post -from nemo_platform_plugin.client.response import NemoHTTPError, NemoResponse +from nemo_platform_plugin.client.errors import NemoHTTPError, NotFoundError +from nemo_platform_plugin.client.response import NemoResponse from pydantic import BaseModel BASE = "http://test:8000" @@ -302,6 +303,7 @@ def test_query_params_all_none_becomes_none() -> None: def test_error_response_extracts_detail() -> None: + """send() raises NemoHTTPError with detail extracted from response body.""" mock_http = MagicMock(spec=httpx.Client) mock_http.request.return_value = httpx.Response( 422, @@ -310,10 +312,9 @@ def test_error_response_extracts_detail() -> None: ) client = NemoClient(base_url=BASE, http_client=mock_http) - resp = client.send(CREATE_ITEM(ItemRequest(name=""))) with pytest.raises(NemoHTTPError) as exc_info: - resp.data() + client.send(CREATE_ITEM(ItemRequest(name=""))) assert exc_info.value.status_code == 422 assert exc_info.value.detail == "Validation failed: name is required" @@ -322,6 +323,7 @@ def test_error_response_extracts_detail() -> None: def test_error_response_fallback_to_text() -> None: + """send() raises NemoHTTPError with raw text when no JSON detail.""" mock_http = MagicMock(spec=httpx.Client) mock_http.request.return_value = httpx.Response( 500, @@ -330,17 +332,16 @@ def test_error_response_fallback_to_text() -> None: ) client = NemoClient(base_url=BASE, http_client=mock_http) - resp = client.send(GET_ITEM(name="x")) with pytest.raises(NemoHTTPError) as exc_info: - resp.data() + client.send(GET_ITEM(name="x")) assert exc_info.value.status_code == 500 assert exc_info.value.detail == "Internal Server Error" -def test_error_response_body_is_none() -> None: - """On error, body should be None (not deserialized as the response type).""" +def test_error_response_raises_specific_subclass() -> None: + """send() raises status-code-specific NemoHTTPError subclass.""" mock_http = MagicMock(spec=httpx.Client) mock_http.request.return_value = httpx.Response( 404, @@ -349,10 +350,12 @@ def test_error_response_body_is_none() -> None: ) client = NemoClient(base_url=BASE, http_client=mock_http) - resp = client.send(GET_ITEM(name="missing")) - assert resp.body is None - assert resp.http_response.status_code == 404 + with pytest.raises(NotFoundError) as exc_info: + client.send(GET_ITEM(name="missing")) + + assert exc_info.value.status_code == 404 + assert exc_info.value.detail == "Not found" # --------------------------------------------------------------------------- @@ -410,3 +413,50 @@ def test_response_carries_prepared_request() -> None: assert resp.request is not None assert resp.request.method == "GET" assert resp.request.path_params == {"name": "alice"} + + +# --------------------------------------------------------------------------- +# Regression tests +# --------------------------------------------------------------------------- + + +def test_send_raises_on_non_2xx() -> None: + """send() must raise immediately on non-2xx.""" + mock_http = MagicMock(spec=httpx.Client) + mock_http.request.return_value = httpx.Response( + 403, + request=httpx.Request("PUT", f"{BASE}/apis/test/upload"), + json={"detail": "Access denied"}, + ) + + client = NemoClient(base_url=BASE, http_client=mock_http) + + with pytest.raises(NemoHTTPError) as exc_info: + client.send(CREATE_ITEM(ItemRequest(name="x"))) + + assert exc_info.value.status_code == 403 + + +def test_query_param_dicts_are_json_serialized() -> None: + """Dict query params must be JSON-serialized, not Python repr.""" + from abc import abstractmethod + + from nemo_platform_plugin.client.endpoint import get + + @get("/apis/test/v2/items") + @abstractmethod + def list_items(*, query_params: dict | None = None) -> ItemResponse: ... + + mock_http = MagicMock(spec=httpx.Client) + mock_http.request.return_value = httpx.Response( + 200, + request=httpx.Request("GET", f"{BASE}/apis/test/v2/items"), + json={"id": 1, "name": "x"}, + ) + + client = NemoClient(base_url=BASE, http_client=mock_http) + client.send(list_items(query_params={"filter": {"name": "test"}})) + + _, kwargs = mock_http.request.call_args + filter_value = kwargs["params"]["filter"] + assert filter_value == '{"name": "test"}', f"Expected JSON string, got: {filter_value}" diff --git a/packages/nemo_platform_plugin/tests/client/test_client_options.py b/packages/nemo_platform_plugin/tests/client/test_client_options.py index 4f85c67391..c45e9bbadb 100644 --- a/packages/nemo_platform_plugin/tests/client/test_client_options.py +++ b/packages/nemo_platform_plugin/tests/client/test_client_options.py @@ -11,7 +11,7 @@ import pytest from nemo_platform_plugin.client.client import AsyncNemoClient, NemoClient from nemo_platform_plugin.client.endpoint import delete, get, post -from nemo_platform_plugin.client.method import method +from nemo_platform_plugin.client.errors import NemoHTTPError from nemo_platform_plugin.client.types import PreparedRequest, RetryPolicy from pydantic import BaseModel @@ -72,97 +72,6 @@ def test_endpoint_without_options_has_none(self) -> None: assert prepared.client_options is None -# --------------------------------------------------------------------------- -# exist_ok: applied via send() -# --------------------------------------------------------------------------- - - -class TestExistOkViaSend: - def test_exist_ok_via_send_swallows_409(self) -> None: - """exist_ok should work when calling client.send() directly.""" - mock_http = MagicMock(spec=httpx.Client) - mock_http.request.return_value = httpx.Response( - 409, - request=httpx.Request("POST", f"{BASE}/apis/test/v2/items"), - json={"id": 1, "name": "alice"}, - ) - - client = NemoClient(base_url=BASE, http_client=mock_http) - resp = client.send(CREATE_ITEM(ItemRequest(name="alice"), exist_ok=True)) - - assert resp.http_response.status_code == 409 - assert resp.body is not None - assert resp.body.name == "alice" - - -# --------------------------------------------------------------------------- -# exist_ok: applied via EndpointMethod -# --------------------------------------------------------------------------- - - -class TestExistOkViaMethod: - def test_exist_ok_true_swallows_409(self) -> None: - mock_http = MagicMock(spec=httpx.Client) - mock_http.request.return_value = httpx.Response( - 409, - request=httpx.Request("POST", f"{BASE}/apis/test/v2/items"), - json={"id": 1, "name": "alice"}, - ) - - class _Methods: - create_item = method(CREATE_ITEM) - - class TestClient(_Methods, NemoClient): - pass - - client = TestClient(base_url=BASE, http_client=mock_http) - resp = client.create_item(body=ItemRequest(name="alice"), exist_ok=True) - - assert resp.http_response.status_code == 409 - assert resp.body is not None - assert resp.body.name == "alice" - - def test_exist_ok_false_returns_409_as_is(self) -> None: - mock_http = MagicMock(spec=httpx.Client) - mock_http.request.return_value = httpx.Response( - 409, - request=httpx.Request("POST", f"{BASE}/apis/test/v2/items"), - json={"detail": "Already exists"}, - ) - - class _Methods: - create_item = method(CREATE_ITEM) - - class TestClient(_Methods, NemoClient): - pass - - client = TestClient(base_url=BASE, http_client=mock_http) - resp = client.create_item(body=ItemRequest(name="alice")) - - assert resp.http_response.status_code == 409 - assert resp.body is None - - def test_exist_ok_true_non_409_passes_through(self) -> None: - mock_http = MagicMock(spec=httpx.Client) - mock_http.request.return_value = httpx.Response( - 201, - request=httpx.Request("POST", f"{BASE}/apis/test/v2/items"), - json={"id": 1, "name": "alice"}, - ) - - class _Methods: - create_item = method(CREATE_ITEM) - - class TestClient(_Methods, NemoClient): - pass - - client = TestClient(base_url=BASE, http_client=mock_http) - resp = client.create_item(body=ItemRequest(name="alice"), exist_ok=True) - - assert resp.http_response.status_code == 201 - assert resp.body.name == "alice" - - # --------------------------------------------------------------------------- # Param validation at decoration time # --------------------------------------------------------------------------- @@ -225,7 +134,7 @@ def test_retry_on_503(self) -> None: assert resp.body.name == "alice" assert mock_http.request.call_count == 2 - def test_retry_exhausted_returns_last_response(self) -> None: + def test_retry_exhausted_raises(self) -> None: mock_http = MagicMock(spec=httpx.Client) mock_http.request.return_value = httpx.Response( 503, @@ -238,9 +147,11 @@ def test_retry_exhausted_returns_last_response(self) -> None: http_client=mock_http, retry=RetryPolicy(max_retries=2, backoff_base=0.0), ) - resp = client.send(GET_ITEM(name="alice")) - assert resp.http_response.status_code == 503 + with pytest.raises(NemoHTTPError) as exc_info: + client.send(GET_ITEM(name="alice")) + + assert exc_info.value.status_code == 503 assert mock_http.request.call_count == 3 def test_no_retry_on_non_retryable_status(self) -> None: @@ -256,9 +167,11 @@ def test_no_retry_on_non_retryable_status(self) -> None: http_client=mock_http, retry=RetryPolicy(max_retries=2, backoff_base=0.0), ) - resp = client.send(GET_ITEM(name="alice")) - assert resp.http_response.status_code == 404 + with pytest.raises(NemoHTTPError) as exc_info: + client.send(GET_ITEM(name="alice")) + + assert exc_info.value.status_code == 404 assert mock_http.request.call_count == 1 def test_retry_on_transport_error(self) -> None: @@ -295,7 +208,9 @@ def test_per_request_retry_overrides_client_default(self) -> None: http_client=mock_http, retry=RetryPolicy(max_retries=5, backoff_base=0.0), ) - client.send(GET_ITEM(name="alice"), retry=RetryPolicy(max_retries=1, backoff_base=0.0)) + + with pytest.raises(NemoHTTPError): + client.send(GET_ITEM(name="alice"), retry=RetryPolicy(max_retries=1, backoff_base=0.0)) assert mock_http.request.call_count == 2 @@ -308,9 +223,11 @@ def test_no_retry_without_policy(self) -> None: ) client = NemoClient(base_url=BASE, http_client=mock_http) - resp = client.send(GET_ITEM(name="alice")) - assert resp.http_response.status_code == 503 + with pytest.raises(NemoHTTPError) as exc_info: + client.send(GET_ITEM(name="alice")) + + assert exc_info.value.status_code == 503 assert mock_http.request.call_count == 1 @@ -319,30 +236,6 @@ def test_no_retry_without_policy(self) -> None: # --------------------------------------------------------------------------- -class TestAsyncExistOk: - @pytest.mark.asyncio - async def test_exist_ok_true_swallows_409_async(self) -> None: - mock_http = AsyncMock(spec=httpx.AsyncClient) - mock_http.request.return_value = httpx.Response( - 409, - request=httpx.Request("POST", f"{BASE}/apis/test/v2/items"), - json={"id": 1, "name": "alice"}, - ) - - class _Methods: - create_item = method(CREATE_ITEM) - - class TestAsyncClient(_Methods, AsyncNemoClient): - pass - - client = TestAsyncClient(base_url=BASE, http_client=mock_http) - resp = await client.create_item(body=ItemRequest(name="alice"), exist_ok=True) - - assert resp.http_response.status_code == 409 - assert resp.body is not None - assert resp.body.name == "alice" - - # --------------------------------------------------------------------------- # Async: RetryPolicy # --------------------------------------------------------------------------- diff --git a/plugins/example-plugin/tests/test_sdk.py b/plugins/example-plugin/tests/test_sdk.py index 0d070e7b48..d73b5ccdef 100644 --- a/plugins/example-plugin/tests/test_sdk.py +++ b/plugins/example-plugin/tests/test_sdk.py @@ -5,6 +5,7 @@ from __future__ import annotations +from contextlib import asynccontextmanager, contextmanager from unittest.mock import AsyncMock, MagicMock import httpx @@ -12,10 +13,12 @@ from nemo_example_plugin.sdk import AsyncExampleClient, ExampleClient from nemo_example_plugin.types import endpoints from nemo_example_plugin.types.payloads import ( + CountRequest, CreateExampleItemRequest, UpdateExampleItemRequest, ) from nemo_platform_plugin.client.client import NemoClient +from nemo_platform_plugin.client.errors import NemoHTTPError BASE = "http://test:8000" WS = "default" @@ -215,3 +218,212 @@ async def test_async_delete_item() -> None: await client.delete_item(name="my-item") mock_http.request.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Binary endpoints — upload_blob / download_blob +# --------------------------------------------------------------------------- + + +def _stream_ctx(resp: httpx.Response): + """Create a sync context manager that yields *resp*.""" + + @contextmanager + def _ctx(*_args, **_kwargs): + yield resp + + return _ctx + + +def _async_stream_ctx(resp: httpx.Response): + """Create an async context manager that yields *resp*.""" + + @asynccontextmanager + async def _ctx(*_args, **_kwargs): + yield resp + + return _ctx + + +def test_sync_upload_blob() -> None: + client, mock_http = _sync_client() + mock_http.request.return_value = _resp(200, {"name": "pic.png", "size": 42}) + + resp = client.upload_blob(name="pic.png", content=b"\x89PNG") + + assert resp.data().name == "pic.png" + assert resp.data().size == 42 + + +def test_sync_download_blob_read() -> None: + client, mock_http = _sync_client() + raw = httpx.Response(200, content=b"file-bytes", request=httpx.Request("GET", BASE)) + mock_http.stream = _stream_ctx(raw) + + resp = client.download_blob(name="pic.png") + data = resp.read() + + assert data == b"file-bytes" + + +def test_sync_download_blob_stream() -> None: + client, mock_http = _sync_client() + raw = httpx.Response(200, content=b"chunk1chunk2", request=httpx.Request("GET", BASE)) + mock_http.stream = _stream_ctx(raw) + + resp = client.download_blob(name="pic.png") + with resp.stream() as chunks: + result = b"".join(chunks) + + assert result == b"chunk1chunk2" + + +@pytest.mark.asyncio +async def test_async_upload_blob() -> None: + client, mock_http = _async_client() + mock_http.request.return_value = _resp(200, {"name": "pic.png", "size": 42}) + + resp = await client.upload_blob(name="pic.png", content=b"\x89PNG") + + assert resp.data().name == "pic.png" + + +@pytest.mark.asyncio +async def test_async_download_blob_read() -> None: + client, mock_http = _async_client() + raw = httpx.Response(200, content=b"file-bytes", request=httpx.Request("GET", BASE)) + mock_http.stream = _async_stream_ctx(raw) + + resp = await client.download_blob(name="pic.png") + data = await resp.read() + + assert data == b"file-bytes" + + +# --------------------------------------------------------------------------- +# Streaming endpoint — count +# --------------------------------------------------------------------------- + + +def test_sync_count_stream() -> None: + client, mock_http = _sync_client() + body = '{"kind":"tick","n":1}\n{"kind":"tick","n":2}\n{"kind":"done","n":null}\n' + raw = httpx.Response(200, content=body.encode(), request=httpx.Request("POST", BASE)) + mock_http.stream = _stream_ctx(raw) + + resp = client.count(body=CountRequest(upto=2)) + with resp.stream() as ticks: + items = list(ticks) + + assert len(items) == 3 + assert items[0].kind == "tick" + assert items[0].n == 1 + assert items[2].kind == "done" + + +@pytest.mark.asyncio +async def test_async_count_stream() -> None: + client, mock_http = _async_client() + body = '{"kind":"tick","n":1}\n{"kind":"done","n":null}\n' + raw = httpx.Response(200, content=body.encode(), request=httpx.Request("POST", BASE)) + mock_http.stream = _async_stream_ctx(raw) + + resp = await client.count(body=CountRequest(upto=1)) + async with resp.stream() as ticks: + items = [t async for t in ticks] + + assert len(items) == 2 + assert items[0].kind == "tick" + assert items[1].kind == "done" + + +# --------------------------------------------------------------------------- +# SSE framing support +# --------------------------------------------------------------------------- + + +def test_sync_stream_sse_framing() -> None: + """SSE data: prefixes are stripped when Content-Type is text/event-stream.""" + client, mock_http = _sync_client() + body = 'data: {"kind":"tick","n":1}\ndata: {"kind":"done","n":null}\n\n' + raw = httpx.Response( + 200, + content=body.encode(), + headers={"content-type": "text/event-stream"}, + request=httpx.Request("POST", BASE), + ) + mock_http.stream = _stream_ctx(raw) + + resp = client.count(body=CountRequest(upto=1)) + with resp.stream() as ticks: + items = list(ticks) + + assert len(items) == 2 + assert items[0].kind == "tick" + assert items[1].kind == "done" + + +def test_sync_stream_sse_skips_non_data_fields() -> None: + """SSE event:, id:, and comment lines are skipped.""" + client, mock_http = _sync_client() + body = 'event: tick\ndata: {"kind":"tick","n":1}\n: comment\nid: 42\ndata: {"kind":"done","n":null}\n\n' + raw = httpx.Response( + 200, + content=body.encode(), + headers={"content-type": "text/event-stream"}, + request=httpx.Request("POST", BASE), + ) + mock_http.stream = _stream_ctx(raw) + + resp = client.count(body=CountRequest(upto=1)) + with resp.stream() as ticks: + items = list(ticks) + + assert len(items) == 2 + assert items[0].kind == "tick" + assert items[1].kind == "done" + + +@pytest.mark.asyncio +async def test_async_stream_sse_framing() -> None: + """Async SSE data: prefixes are stripped.""" + client, mock_http = _async_client() + body = 'data: {"kind":"tick","n":1}\ndata: {"kind":"done","n":null}\n\n' + raw = httpx.Response( + 200, + content=body.encode(), + headers={"content-type": "text/event-stream"}, + request=httpx.Request("POST", BASE), + ) + mock_http.stream = _async_stream_ctx(raw) + + resp = await client.count(body=CountRequest(upto=1)) + async with resp.stream() as ticks: + items = [t async for t in ticks] + + assert len(items) == 2 + assert items[0].kind == "tick" + assert items[1].kind == "done" + + +# --------------------------------------------------------------------------- +# Error detail extraction from streaming responses +# --------------------------------------------------------------------------- + + +def test_binary_read_error_has_detail() -> None: + """Binary read() on error response should extract JSON detail.""" + client, mock_http = _sync_client() + raw = httpx.Response( + 404, + content=b'{"detail": "File not found"}', + request=httpx.Request("GET", BASE), + ) + mock_http.stream = _stream_ctx(raw) + + resp = client.download_blob(name="missing.png") + with pytest.raises(NemoHTTPError) as exc_info: + resp.read() + + assert exc_info.value.status_code == 404 + assert exc_info.value.detail == "File not found"