diff --git a/README.md b/README.md index e2b02c4..f48a07a 100644 --- a/README.md +++ b/README.md @@ -74,9 +74,10 @@ headers are typed. ## 3. Full — the `Routeplane` client with rich metadata -`Routeplane` subclasses `openai.OpenAI`, so everything works exactly as before — -but it wires up auth for you, lets you set default routing once, and can parse -the gateway's response headers into a typed `RouteplaneMeta`. +`Routeplane` subclasses `openai.OpenAI` and keeps its resource/request API. It +wires up auth, lets you set default routing once, and parses the gateway's +response headers into a typed `RouteplaneMeta`. Chat response tier handling is +provider-neutral at runtime; see the served-tier contract below. ```python from routeplane import Routeplane @@ -213,6 +214,65 @@ parses the gateway's `x-routeplane-*` response headers: `provider`, `trace_id`, `request_id`, `cache`, `guardrails`, `hedged`, `shed`, `budget_remaining`, `budget_warning`, `compliance_warning`, `pii_masked`, `idempotent_replayed`. +### Actual served tier + +`create_with_meta()` returns a `RouteplaneChatCompletion`, and `stream_with_meta()` +yields `RouteplaneChatCompletionChunk` objects, on both sync and async clients. +These exported models retain the installed OpenAI response fields and validation +while widening only returned `service_tier` to an optional strict string. Native +labels such as `standard`, `priority`, `batch`, and future provider labels keep +their exact spelling. Missing or `null` tier evidence remains `None`; serialization +with `exclude_unset=True` preserves omission versus explicit `null`. Non-string, +non-null tiers raise `APIResponseValidationError` even in default tolerant mode. +Unrelated response fields continue to follow the vendor's default or opt-in +`_strict_response_validation=True` behavior. + +```python +from routeplane import Routeplane, RouteplaneChatCompletion + +client = Routeplane(api_key="rp_your_gateway_key") +completion, meta = client.create_with_meta( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) +assert isinstance(completion, RouteplaneChatCompletion) +print(completion.service_tier) # provider-reported evidence, possibly None + +with client.stream_with_meta( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) as stream: + for chunk in stream: + # Continue past a finish_reason: tier-only metadata may arrive afterward. + # choices may be empty, and usage may be omitted on that metadata chunk. + if chunk.service_tier is not None: + print(chunk.service_tier) +``` + +The returned tier is never inferred from a requested tier, renamed, or treated as +proof of a free response, tariff eligibility, complete cost, or billing authority. +Streaming keeps the vendor SSE decoder and `[DONE]` termination; no aggregation or +synthetic usage is added. + +The inherited `client.chat.completions` resource signatures still expose the +**OpenAI vendor model types**. Routeplane's exact-model response-processing hook +also uses the neutral models at runtime for those calls, including raw responses, +but it does not widen the inherited static facade. Use the `*_with_meta` helpers +for a typed neutral result. On OpenAI versions that support custom raw parsing, +an explicit neutral model is also available: + +```python +raw = client.chat.completions.with_raw_response.create( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) +completion = raw.parse(to=RouteplaneChatCompletion) +``` + +OpenAI 1.0.0 has no `parse(to=...)` argument. The helpers and the runtime hook +support that advertised dependency floor using its existing response processor; +`raw.parse()` receives the neutral runtime model there too. A stock `openai.OpenAI` +client and custom response-model targets retain their own validation behavior. +Stream cleanup follows the installed vendor version; OpenAI 1.0.0 itself has no +`Stream.close()` or `AsyncStream.close()` method. + ## Legacy feedback Use the gateway-generated request ID from response metadata, not the provider's diff --git a/src/routeplane/__init__.py b/src/routeplane/__init__.py index ccd9963..679f14b 100644 --- a/src/routeplane/__init__.py +++ b/src/routeplane/__init__.py @@ -45,12 +45,15 @@ TimeseriesData, UsageData, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk __all__ = [ "Routeplane", "AsyncRouteplane", "RouteplaneStream", "AsyncRouteplaneStream", + "RouteplaneChatCompletion", + "RouteplaneChatCompletionChunk", "headers", "RouteplaneMeta", "RouteplaneRateLimits", diff --git a/src/routeplane/async_client.py b/src/routeplane/async_client.py index bff7a78..9be52e1 100644 --- a/src/routeplane/async_client.py +++ b/src/routeplane/async_client.py @@ -16,7 +16,7 @@ import httpx import openai -from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai._types import ResponseT from ._streaming import AsyncRouteplaneStream from .client import DEFAULT_BASE_URL @@ -35,6 +35,7 @@ ResidencyResource, StatusResource, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk, _response_type __all__ = ["AsyncRouteplane"] @@ -139,7 +140,19 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: """ return RouteplaneMeta.from_headers(headers) - async def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMeta]: + def _process_response_data( + self, *, data: object, cast_to: type[ResponseT], response: Any + ) -> ResponseT: + # SDK versions use httpx or httpx2; forward the opaque response unchanged. + return super()._process_response_data( + data=data, + cast_to=_response_type(cast_to=cast_to, data=data, response=response), + response=response, + ) + + async def create_with_meta( + self, **kwargs: Any + ) -> Tuple[RouteplaneChatCompletion, RouteplaneMeta]: """Chat completion that also returns the gateway :class:`RouteplaneMeta`. :: @@ -147,10 +160,12 @@ async def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, Routepl completion, meta = await client.create_with_meta(model="gpt-4o", messages=[...]) """ raw = await self.chat.completions.with_raw_response.create(**kwargs) - completion = cast(ChatCompletion, raw.parse()) + completion = cast(RouteplaneChatCompletion, raw.parse()) return completion, RouteplaneMeta.from_headers(raw.headers) - async def stream_with_meta(self, **kwargs: Any) -> AsyncRouteplaneStream[ChatCompletionChunk]: + async def stream_with_meta( + self, **kwargs: Any + ) -> AsyncRouteplaneStream[RouteplaneChatCompletionChunk]: """Streaming chat completion that also exposes ``meta`` on the stream. :: diff --git a/src/routeplane/client.py b/src/routeplane/client.py index cf84681..2d1717f 100644 --- a/src/routeplane/client.py +++ b/src/routeplane/client.py @@ -1,7 +1,9 @@ """The :class:`Routeplane` client — a thin subclass of ``openai.OpenAI``. -Everything the OpenAI SDK can do works unchanged (``client.chat.completions``, -``client.embeddings``, streaming, retries). The subclass adds three things: +The OpenAI resource API remains available (``client.chat.completions``, +``client.embeddings``, streaming, retries). Chat responses use provider-neutral +returned-tier models at runtime; inherited resource annotations stay vendor-owned. +The subclass adds three things: 1. gateway auth + default ``x-routeplane-*`` routing headers, so callers don't repeat them on every request; @@ -17,7 +19,7 @@ import httpx import openai -from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai._types import ResponseT from ._streaming import RouteplaneStream from .headers import headers as build_headers @@ -35,6 +37,7 @@ ResidencyResource, StatusResource, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk, _response_type __all__ = ["Routeplane"] @@ -153,7 +156,17 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: """ return RouteplaneMeta.from_headers(headers) - def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMeta]: + def _process_response_data( + self, *, data: object, cast_to: type[ResponseT], response: Any + ) -> ResponseT: + # SDK versions use httpx or httpx2; forward the opaque response unchanged. + return super()._process_response_data( + data=data, + cast_to=_response_type(cast_to=cast_to, data=data, response=response), + response=response, + ) + + def create_with_meta(self, **kwargs: Any) -> Tuple[RouteplaneChatCompletion, RouteplaneMeta]: """Chat completion that also returns the gateway :class:`RouteplaneMeta`. :: @@ -162,10 +175,10 @@ def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMet print(meta.provider, meta.cache) """ raw = self.chat.completions.with_raw_response.create(**kwargs) - completion = cast(ChatCompletion, raw.parse()) + completion = cast(RouteplaneChatCompletion, raw.parse()) return completion, RouteplaneMeta.from_headers(raw.headers) - def stream_with_meta(self, **kwargs: Any) -> RouteplaneStream[ChatCompletionChunk]: + def stream_with_meta(self, **kwargs: Any) -> RouteplaneStream[RouteplaneChatCompletionChunk]: """Streaming chat completion that also exposes ``meta`` on the stream. ``meta`` is populated from the response headers, which arrive before the diff --git a/src/routeplane/responses.py b/src/routeplane/responses.py new file mode 100644 index 0000000..75bc47e --- /dev/null +++ b/src/routeplane/responses.py @@ -0,0 +1,67 @@ +"""OpenAI-shaped responses with provider-neutral actual served-tier evidence.""" + +from __future__ import annotations + +from typing import Any, Optional, TypeVar, cast + +from openai import APIResponseValidationError +from openai.types.chat import ChatCompletion, ChatCompletionChunk +from pydantic import StrictStr + +__all__ = ["RouteplaneChatCompletion", "RouteplaneChatCompletionChunk"] + + +class RouteplaneChatCompletion(ChatCompletion): + """A completion whose returned tier is an exact provider string or ``None``. + + All other fields and validation come from the installed OpenAI model. The + label is evidence of what served the request, not a tariff or cost claim. + """ + + # Intentionally widen the vendor literal union at this one wire-contract seam. + # The 1.0.0 model predates this field, so the assignment ignore is unused there. + service_tier: Optional[StrictStr] = None # type: ignore[assignment, unused-ignore] + + +class RouteplaneChatCompletionChunk(ChatCompletionChunk): + """A stream chunk with the same provider-neutral returned-tier contract.""" + + # Same deliberate widening, including the dependency floor without this field. + service_tier: Optional[StrictStr] = None # type: ignore[assignment, unused-ignore] + + +_T = TypeVar("_T") + + +def _response_type(*, cast_to: type[_T], data: object, response: Any) -> type[_T]: + """Select the neutral model and check only tier even in tolerant mode. + + The client's existing response processor still controls validation of every + other field. Exact class matching leaves custom models and other APIs alone. + This hook is present at the supported OpenAI 1.0.0 dependency floor, where + raw-response ``parse(to=...)`` is not available yet. + + The opaque vendor response is forwarded unchanged: supported SDK versions + use unrelated httpx/httpx2 classes and export no shared response type. + Data, model selection, and returned models retain their typed contracts. + """ + neutral: type[object] + if cast_to is ChatCompletion: + neutral = RouteplaneChatCompletion + elif cast_to is ChatCompletionChunk: + neutral = RouteplaneChatCompletionChunk + elif cast_to in (RouteplaneChatCompletion, RouteplaneChatCompletionChunk): + neutral = cast_to + else: + return cast_to + + if isinstance(data, dict): + tier = data.get("service_tier") + if tier is not None and not isinstance(tier, str): + raise APIResponseValidationError( + response=response, + body=data, + message="Expected response service_tier to be a string or null.", + ) + + return cast("type[_T]", neutral) diff --git a/tests/test_served_tier.py b/tests/test_served_tier.py new file mode 100644 index 0000000..9b3c932 --- /dev/null +++ b/tests/test_served_tier.py @@ -0,0 +1,597 @@ +"""Exercise actual vendor decoding, including strict mode and trailing SSE metadata.""" + +import inspect +import json +import subprocess +import sys +from pathlib import Path + +import httpx +import openai +import pytest +import respx +from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai.types.create_embedding_response import CreateEmbeddingResponse +from pydantic import ValidationError + +from routeplane import ( + AsyncRouteplane, + Routeplane, + RouteplaneChatCompletion, + RouteplaneChatCompletionChunk, +) + +BASE = "https://api.routeplane.ai/v1" +CHAT = f"{BASE}/chat/completions" +OMITTED = object() +TIERS = [OMITTED, None, "standard", "priority", "batch", "default", "future-tier", " Native "] +USAGE = {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5} +META = { + "x-routeplane-provider": "openrouter", + "x-routeplane-request-id": "rp_request", + "x-routeplane-cache": "bypass", +} +REQUEST = { + "model": "model", + "messages": [{"role": "user", "content": "hi"}], + "temperature": 0.25, + "max_tokens": 16, + # extra_body works even at OpenAI 1.0.0, before its named service_tier input. + "extra_body": {"service_tier": "priority", "provider_option": {"keep": True}}, + "extra_headers": {"x-routeplane-provider": "openrouter", "x-custom": "kept"}, +} + + +def completion(tier=OMITTED): + body = { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1, + "model": "model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "hello"}, + "finish_reason": "stop", + } + ], + "usage": USAGE, + "provider_extension": {"keep": True}, + } + if tier is not OMITTED: + body["service_tier"] = tier + return body + + +def chunk(content=None, finish=None): + return { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 1, + "model": "model", + "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": finish}], + } + + +def frames(tier=OMITTED): + # The provider reports its tier only AFTER the finish chunk, without usage. + finish = chunk(None, "stop") + finish["usage"] = USAGE + metadata = {**chunk(), "choices": [], "provider_extension": {"keep": True}} + if tier is not OMITTED: + metadata["service_tier"] = tier + return [chunk("hello"), finish, metadata] + + +def response(tier, streaming): + if not streaming: + return httpx.Response(200, json=completion(tier), headers=META) + body = "".join(f"data: {json.dumps(frame)}\n\n" for frame in frames(tier)) + # A frame after [DONE] must never become another completion chunk. + body += f"data: [DONE]\n\ndata: {json.dumps(chunk('ignored'))}\n\n" + return httpx.Response( + 200, content=body.encode(), headers={"content-type": "text/event-stream", **META} + ) + + +def assert_request(route, streaming): + sent = route.calls.last.request + assert sent.headers["x-routeplane-api-key"] == "rp_test" + assert sent.headers["authorization"] == "Bearer rp_test" + assert sent.headers["x-routeplane-provider"] == "openrouter" + assert sent.headers["x-custom"] == "kept" + expected = { + "model": "model", + "messages": REQUEST["messages"], + "temperature": 0.25, + "max_tokens": 16, + **REQUEST["extra_body"], + } + if streaming: + expected["stream"] = True + assert json.loads(sent.content) == expected + + +def assert_result(parsed, tier, streaming, strict): + if streaming: + assert len(parsed) == 3 + assert all(isinstance(c, RouteplaneChatCompletionChunk) for c in parsed) + assert parsed[0].choices[0].delta.content == "hello" + assert parsed[0].service_tier is None # never infer the requested priority + assert parsed[1].choices[0].finish_reason == "stop" + usage = parsed[1].usage + assert (usage if isinstance(usage, dict) else usage.dict(exclude_unset=True)) == USAGE + result = parsed[-1] + expected = frames(tier)[-1] + assert result.choices == [] + assert "usage" not in result.dict(exclude_unset=True) + else: + assert isinstance(parsed, RouteplaneChatCompletion) + result = parsed + expected = completion(tier) + assert result.usage.total_tokens == 5 + assert result.usage.prompt_tokens == 3 + assert result.usage.completion_tokens == 2 + assert result.service_tier == (None if tier is OMITTED else tier) + assert result.provider_extension == {"keep": True} + # Preserve the installed vendor's serialization of all unrelated fields. + # OpenAI 1.0.0/Pydantic 1's tolerant construct omits extras under exclude_unset, + # though it retains their attributes. Do not change that vendor behavior. + native = ChatCompletionChunk if streaming else ChatCompletion + base_body = {k: v for k, v in expected.items() if k != "service_tier"} + vendor = native(**base_body) if strict else native.construct(**base_body) + expected_dump = vendor.dict(exclude_unset=True) + if tier is not OMITTED: + expected_dump["service_tier"] = tier + # Omission vs null and the exact native spelling survive in every mode. + assert result.dict(exclude_unset=True) == expected_dump + + +@pytest.mark.parametrize("tier", TIERS) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_sync_actual_served_tier(tier, strict, streaming): + route = respx.post(CHAT).mock(return_value=response(tier, streaming)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + provider="openai", + http_client=httpx.Client(), + _strict_response_validation=strict, + ) as client: + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + meta = stream.meta + parsed = list(stream) + else: + parsed, meta = client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert meta.request_id == "rp_request" + assert meta.cache == "bypass" + assert_result(parsed, tier, streaming, strict) + assert_request(route, streaming) + + +@pytest.mark.parametrize("tier", TIERS) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_actual_served_tier(tier, strict, streaming): + route = respx.post(CHAT).mock(return_value=response(tier, streaming)) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + provider="openai", + http_client=httpx.AsyncClient(), + _strict_response_validation=strict, + ) as client: + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + meta = stream.meta + parsed = [c async for c in stream] + else: + parsed, meta = await client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert meta.request_id == "rp_request" + assert meta.cache == "bypass" + assert_result(parsed, tier, streaming, strict) + assert_request(route, streaming) + + +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}]) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_sync_rejects_invalid_tier(tier, strict, streaming): + respx.post(CHAT).mock(return_value=response(tier, streaming)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=strict, + ) as client: + with pytest.raises(openai.APIResponseValidationError, match="service_tier"): + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + list(stream) + else: + client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}]) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_rejects_invalid_tier(tier, strict, streaming): + respx.post(CHAT).mock(return_value=response(tier, streaming)) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + _strict_response_validation=strict, + ) as client: + with pytest.raises(openai.APIResponseValidationError, match="service_tier"): + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + [c async for c in stream] + else: + await client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("model", [RouteplaneChatCompletion, RouteplaneChatCompletionChunk]) +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}, b"standard"]) +def test_exported_models_validate_tier_without_coercion(model, tier): + body = completion() if model is RouteplaneChatCompletion else chunk() + body["service_tier"] = tier + with pytest.raises(ValidationError): + model(**body) + + +@pytest.mark.parametrize( + "neutral,native", + [ + (RouteplaneChatCompletion, ChatCompletion), + (RouteplaneChatCompletionChunk, ChatCompletionChunk), + ], +) +def test_neutral_models_only_change_the_returned_tier_field(neutral, native): + field_attribute = "model_fields" if hasattr(native, "model_fields") else "__fields__" + native_fields = getattr(native, field_attribute) + neutral_fields = getattr(neutral, field_attribute) + assert set(neutral_fields) == set(native_fields) | {"service_tier"} + for name, field in native_fields.items(): + if name != "service_tier": + inherited = neutral_fields[name] + assert inherited.annotation == field.annotation + assert inherited.default == field.default + assert inherited.alias == field.alias + + +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_strict_validation_of_other_fields_is_retained(streaming): + body = chunk() if streaming else completion("standard") + body["created"] = {"invalid": True} + mock_response = ( + httpx.Response( + 200, + content=f"data: {json.dumps(body)}\n\n".encode(), + headers={"content-type": "text/event-stream"}, + ) + if streaming + else httpx.Response(200, json=body) + ) + respx.post(CHAT).mock(return_value=mock_response) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=True, + ) as client: + # OpenAI 1.0.0 exposes Pydantic's validation error directly for SSE; + # recent versions wrap it. Preserve that vendor exception boundary. + with pytest.raises((openai.APIResponseValidationError, ValidationError)): + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + list(stream) + else: + client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_strict_validation_of_other_fields_is_retained(streaming): + body = chunk() if streaming else completion("standard") + body["created"] = {"invalid": True} + mock_response = ( + httpx.Response( + 200, + content=f"data: {json.dumps(body)}\n\n".encode(), + headers={"content-type": "text/event-stream"}, + ) + if streaming + else httpx.Response(200, json=body) + ) + respx.post(CHAT).mock(return_value=mock_response) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + _strict_response_validation=True, + ) as client: + with pytest.raises((openai.APIResponseValidationError, ValidationError)): + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + [c async for c in stream] + else: + await client.create_with_meta(**REQUEST) + + +@respx.mock +def test_raw_and_inherited_resource_runtime_boundary(): + respx.post(CHAT).mock(return_value=response("batch", False)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=True, + ) as client: + direct = client.chat.completions.create(**REQUEST) + assert isinstance(direct, RouteplaneChatCompletion) + assert direct.service_tier == "batch" + raw = client.chat.completions.with_raw_response.create(**REQUEST) + # Works at the advertised floor; the return annotation remains vendor-owned. + assert isinstance(raw.parse(), RouteplaneChatCompletion) + if "to" in inspect.signature(raw.parse).parameters: + assert raw.parse(to=RouteplaneChatCompletion).service_tier == "batch" + + +@respx.mock +def test_retries_stay_on_the_vendor_transport(): + route = respx.post(CHAT).mock(side_effect=[httpx.Response(500), response("standard", False)]) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + max_retries=1, + ) as client: + parsed, _ = client.create_with_meta(**REQUEST) + assert parsed.service_tier == "standard" + assert route.call_count == 2 + assert_request(route, False) + + +@respx.mock +async def test_async_retries_stay_on_the_vendor_transport(): + route = respx.post(CHAT).mock(side_effect=[httpx.Response(500), response("standard", False)]) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + max_retries=1, + ) as client: + parsed, _ = await client.create_with_meta(**REQUEST) + assert parsed.service_tier == "standard" + assert route.call_count == 2 + assert_request(route, False) + + +class SyncFrames(httpx.SyncByteStream): + def __init__(self): + self.read_count = 0 + self.closed = False + + def __iter__(self): + for frame in frames("batch"): + self.read_count += 1 + yield f"data: {json.dumps(frame)}\n\n".encode() + yield b"data: [DONE]\n\n" + + def close(self): + self.closed = True + + +class AsyncFrames(httpx.AsyncByteStream): + def __init__(self): + self.read_count = 0 + self.closed = False + + async def __aiter__(self): + for frame in frames("batch"): + self.read_count += 1 + yield f"data: {json.dumps(frame)}\n\n".encode() + yield b"data: [DONE]\n\n" + + async def aclose(self): + self.closed = True + + +def test_sync_stream_stays_incremental_and_closes_on_exit(): + wire = SyncFrames() + transport = httpx.MockTransport( + lambda request: httpx.Response( + 200, stream=wire, headers={"content-type": "text/event-stream", **META} + ) + ) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(transport=transport), + ) as client: + with client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + assert wire.read_count == 0 + assert next(stream).choices[0].delta.content == "hello" + assert wire.read_count == 1 + # Earliest OpenAI's Stream had no close method; that vendor limitation + # predates this change. No new eager draining is introduced at the floor. + if hasattr(openai.Stream, "close"): + assert wire.closed + assert wire.read_count == 1 + + +async def test_async_stream_stays_incremental_and_closes_on_exit(): + wire = AsyncFrames() + transport = httpx.MockTransport( + lambda request: httpx.Response( + 200, stream=wire, headers={"content-type": "text/event-stream", **META} + ) + ) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(transport=transport), + ) as client: + async with await client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + assert wire.read_count == 0 + assert (await stream.__anext__()).choices[0].delta.content == "hello" + assert wire.read_count == 1 + if hasattr(openai.AsyncStream, "close"): + assert wire.closed + assert wire.read_count == 1 + + +class CustomCompletion(ChatCompletion): + provider_marker: str + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("client_type", [Routeplane, AsyncRouteplane]) +def test_custom_and_other_response_targets_remain_untouched(client_type, strict): + client = client_type(api_key="rp_test", _strict_response_validation=strict) + wire_response = httpx.Response(200, request=httpx.Request("POST", CHAT)) + custom_body = {**completion(), "provider_marker": "kept"} + custom = client._process_response_data( + data=custom_body, + cast_to=CustomCompletion, + response=wire_response, + ) + assert type(custom) is CustomCompletion + assert custom.provider_marker == "kept" + embedding_body = { + "object": "list", + "model": "model", + "data": [{"object": "embedding", "index": 0, "embedding": [0.5, 0.25]}], + "usage": {"prompt_tokens": 3, "total_tokens": 3}, + "service_tier": 123, # An extra on another API must not acquire chat validation. + } + embeddings = client._process_response_data( + data=embedding_body, + cast_to=CreateEmbeddingResponse, + response=wire_response, + ) + assert type(embeddings) is CreateEmbeddingResponse + assert embeddings.data[0].embedding == [0.5, 0.25] + assert embeddings.service_tier == 123 + # No request is sent; close both resource clients deterministically. + if isinstance(client, Routeplane): + client.close() + else: + import asyncio + + asyncio.run(client.close()) + + +def test_served_tier_consumer_type_contract(): + """The existing pytest CI gate must also execute the consumer type fixture.""" + root = Path(__file__).resolve().parents[1] + checked = subprocess.run( + [ + sys.executable, + "-m", + "mypy", + "--strict", + "--cache-dir", + "/dev/null", + "src/", + "tests/typing/served_tier.py", + ], + cwd=root, + capture_output=True, + text=True, + timeout=60, + ) + assert checked.returncode == 0, checked.stdout + checked.stderr + + +def vendor_http_backend(): + """Use the installed SDK's actual transport, including recent httpx2.""" + # Read the trusted SDK binding without evaluating postponed annotations, + # which can contain union syntax Python 3.9 cannot evaluate. + sdk_globals = openai.APIResponseValidationError.__init__.__globals__ + return sdk_globals["httpx2"] if "httpx2" in sdk_globals else sdk_globals["httpx"] + + +def vendor_wire_response(backend, tier, streaming): + if not streaming: + return backend.Response(200, json=completion(tier), headers=META) + body = "".join(f"data: {json.dumps(frame)}\n\n" for frame in frames(tier)) + return backend.Response( + 200, + content=(body + "data: [DONE]\n\n").encode(), + headers={"content-type": "text/event-stream", **META}, + ) + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_sync_with_native_vendor_transport(strict, streaming): + backend = vendor_http_backend() + tier = "batch" if streaming else "standard" + transport = backend.MockTransport( + lambda request: vendor_wire_response(backend, tier, streaming) + ) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.Client(transport=transport), + _strict_response_validation=strict, + ) as client: + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + parsed = list(stream) + else: + parsed, meta = client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert_result(parsed, tier, streaming, strict) + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +async def test_async_with_native_vendor_transport(strict, streaming): + backend = vendor_http_backend() + tier = "batch" if streaming else "standard" + transport = backend.MockTransport( + lambda request: vendor_wire_response(backend, tier, streaming) + ) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.AsyncClient(transport=transport), + _strict_response_validation=strict, + ) as client: + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + parsed = [c async for c in stream] + else: + parsed, meta = await client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert_result(parsed, tier, streaming, strict) + + +def test_invalid_tier_keeps_the_native_vendor_response_object(): + backend = vendor_http_backend() + wire = vendor_wire_response(backend, 123, False) + transport = backend.MockTransport(lambda request: wire) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.Client(transport=transport), + ) as client: + with pytest.raises(openai.APIResponseValidationError) as failure: + client.create_with_meta(**REQUEST) + assert failure.value.response is wire diff --git a/tests/typing/served_tier.py b/tests/typing/served_tier.py new file mode 100644 index 0000000..dcf0212 --- /dev/null +++ b/tests/typing/served_tier.py @@ -0,0 +1,57 @@ +"""Static consumer contract: run ``mypy --strict src/ tests/typing/served_tier.py``.""" + +from typing import Optional, Tuple + +from typing_extensions import assert_type + +from routeplane import ( + AsyncRouteplane, + AsyncRouteplaneStream, + Routeplane, + RouteplaneChatCompletion, + RouteplaneChatCompletionChunk, + RouteplaneMeta, + RouteplaneStream, +) + + +def returned_tier(completion: RouteplaneChatCompletion) -> Optional[str]: + assert_type(completion.service_tier, Optional[str]) + completion.service_tier = "unknown-native" + completion.service_tier = None + completion.service_tier = 123 # type: ignore[assignment] + return completion.service_tier + + +def chunk_tier(chunk: RouteplaneChatCompletionChunk) -> Optional[str]: + assert_type(chunk.service_tier, Optional[str]) + chunk.service_tier = "batch" + chunk.service_tier = None + chunk.service_tier = False # type: ignore[assignment] + return chunk.service_tier + + +def sync_helpers(client: Routeplane) -> None: + result = client.create_with_meta(model="model", messages=[]) + assert_type(result, Tuple[RouteplaneChatCompletion, RouteplaneMeta]) + tier: Optional[str] = result[0].service_tier + stream = client.stream_with_meta(model="model", messages=[]) + assert_type(stream, RouteplaneStream[RouteplaneChatCompletionChunk]) + for chunk in stream: + assert_type(chunk, RouteplaneChatCompletionChunk) + assert_type(chunk.service_tier, Optional[str]) + tier = chunk.service_tier + print(tier) + + +async def async_helpers(client: AsyncRouteplane) -> None: + result = await client.create_with_meta(model="model", messages=[]) + assert_type(result, Tuple[RouteplaneChatCompletion, RouteplaneMeta]) + tier: Optional[str] = result[0].service_tier + stream = await client.stream_with_meta(model="model", messages=[]) + assert_type(stream, AsyncRouteplaneStream[RouteplaneChatCompletionChunk]) + async for chunk in stream: + assert_type(chunk, RouteplaneChatCompletionChunk) + assert_type(chunk.service_tier, Optional[str]) + tier = chunk.service_tier + print(tier)