test: passthrough endpoints stream/non-stream e2e

tests/e2e/llm_translation/LLM_TRANSLATION_COVERAGE_MATRIX.md

@@ -0,0 +1,84 @@
+# LLM Translation Test Coverage Matrix
+
+Scope: the proxy's two translation surfaces, end to end against a live proxy.
+
+1. **Passthrough** - the client speaks the provider's NATIVE API (Gemini
+   `generateContent`, Anthropic `/v1/messages`); the proxy forwards it and still
+   logs a costed `SpendLogs` row (`call_type="pass_through_endpoint"`). Routes:
+   `/gemini`, `/anthropic`, `/vertex_ai`, `/openai`, `/bedrock`, `/cohere`,
+   `/mistral`, `/vllm`.
+2. **Non-passthrough** - the client speaks OpenAI format
+   (`/chat/completions`, `/embeddings`); litellm translates to/from the provider.
+
+The two axes that must work in production for each: **passthrough vs
+non-passthrough** and **streaming vs non-streaming**, with **cost logged** and
+**tool calls** working in every cell.
+
+Companion: live suite `test_passthrough_e2e.py` (this directory). The
+non-passthrough chat/embedding cells are exercised by `../spend_tracking/`.
+
+Levels: `live` real provider + proxy + SpendLogs row; `unit` mocked.
+Status: `covered` / `partial` / `gap`.
+
+---
+
+## Passthrough endpoints (native provider format)
+
+| Provider | Non-streaming | Streaming | Tool calls | Cost logged | Status |
+|----------|---------------|-----------|------------|-------------|--------|
+| Gemini (`/gemini/v1beta/models/{m}:generateContent` / `:streamGenerateContent`) | live | live | live | live | **covered** |
+| Anthropic (`/anthropic/v1/messages`) | live | live | live | live | **covered** |
+| Vertex AI (`/vertex_ai/...`) | - | - | - | - | gap (gcloud auth) |
+| OpenAI / Bedrock / Cohere / Mistral / VLLM | - | - | - | - | gap |
+
+Each covered cell asserts: `call_type == "pass_through_endpoint"`, `spend > 0`,
+`status == "success"`, correct `custom_llm_provider`/`model`, row correlated by the
+`x-litellm-call-id` header. Gemini non-streaming also pins `request_tags`
+propagation; streaming pins `chunks > 0` then a costed row; tool tests assert the
+provider emitted a tool call (`functionCall` / `tool_use`) and it was costed.
+
+Cost on passthrough is computed in the success handler by transforming the native
+response to a `ModelResponse` and calling `litellm.completion_cost()`; for
+streaming, chunks are buffered and costed after the stream ends. This is the path
+most likely to silently break and the one a mock can't prove works.
+
+## Non-passthrough endpoints (OpenAI-compatible translation)
+
+| Modality | Non-streaming | Streaming | Tool calls | Cost logged | Status |
+|----------|---------------|-----------|------------|-------------|--------|
+| Chat | live (spend suite) | live (spend suite) | gap | live | partial |
+| Embeddings | live (spend suite) | n/a | n/a | live | covered |
+| Responses / image / audio / rerank / realtime | - | - | - | - | gap |
+
+## This suite's files
+
+| Test | Cell |
+|------|------|
+| `test_gemini_passthrough_nonstreaming_logs_cost` | gemini native, non-stream, cost + tags |
+| `test_gemini_passthrough_streaming_logs_cost` | gemini native, stream, cost |
+| `test_gemini_passthrough_tool_call_logs_cost` | gemini native, tool call, cost |
+| `test_anthropic_passthrough_nonstreaming_logs_cost` | anthropic native, non-stream, cost |
+| `test_anthropic_passthrough_streaming_logs_cost` | anthropic native, stream, cost |
+| `test_anthropic_passthrough_tool_call_logs_cost` | anthropic native, tool call, cost |
+
+## Gaps
+
+- Vertex / OpenAI / Bedrock / Cohere passthrough (same shape; add once the
+  provider credential is configured; Vertex is closest - route exists, auth stale).
+- Non-passthrough tool calls over `/chat/completions` end to end with cost.
+- Image / audio / rerank / responses / realtime translation + cost.
+- Streaming cost-injection (`include_cost_in_streaming_usage`); passthrough on
+  client disconnect (partial-usage logging).
+
+## Adding a provider/modality
+
+Extend `PassthroughClient` with the native call (it inherits keys, cleanup, and
+SpendLogs polling from `ProxyClient`), then add a test that calls it,
+`require_successful_call(result)`, and `_costed_row(...)`.
+
+## Timing
+
+Passthrough spend is logged asynchronously after the response and lands on the
+`proxy_batch_write_at` (~60s) cycle, so cost assertions poll
+`/spend/logs?request_id=<x-litellm-call-id>` to a deadline. Streaming cost is only
+known after the stream is fully consumed.

tests/e2e/llm_translation/conftest.py

@@ -0,0 +1,16 @@
+"""LLM-translation suite's `client` fixture.
+
+The shared lifecycle (resources/scoped_key), proxy liveness skip, and e2e marker
+live in the parent tests/e2e/conftest.py. PassthroughClient (via ProxyClient)
+exposes the shared Gateway, so the `resources` fixture cleans up keys this suite
+creates.
+"""
+
+import pytest
+
+from passthrough_client import PassthroughClient, build_client
+
+
+@pytest.fixture(scope="session")
+def client() -> PassthroughClient:
+    return build_client()

tests/e2e/llm_translation/passthrough_client.py

@@ -0,0 +1,151 @@
+"""Client for LLM-translation e2e tests over the proxy's passthrough endpoints.
+
+Extends the shared ProxyClient with native provider passthrough calls. A
+passthrough request is sent in the PROVIDER's native format (Gemini
+generateContent, Anthropic /v1/messages) to the proxy, which forwards it to the
+provider and still logs a SpendLogs row (call_type="pass_through_endpoint"). The
+litellm virtual key is passed as the provider key; the proxy swaps in the real
+env credential. SpendLogs.request_id == the x-litellm-call-id response header.
+"""
+
+from dataclasses import dataclass
+from typing import Dict, Iterator, List, Optional, cast
+
+import requests
+
+from e2e_config import (
+    MASTER_KEY,
+    POLL_INTERVAL,
+    POLL_TIMEOUT,
+    PROXY_BASE_URL,
+    REQUEST_TIMEOUT,
+)
+from proxy_client import ProxyClient
+
+Tools = List[Dict[str, object]]
+
+
+@dataclass(frozen=True, slots=True)
+class PassthroughResult:
+    """Outcome of a native passthrough call. ``call_id`` correlates to the row."""
+
+    status_code: int
+    call_id: Optional[str]  # x-litellm-call-id -> SpendLogs.request_id
+    body: str
+    chunks: int = 0  # number of streamed events (0 for non-streaming)
+
+    @property
+    def ok(self) -> bool:
+        return 200 <= self.status_code < 300
+
+
+def _tag_header(tags: Optional[List[str]]) -> Dict[str, str]:
+    return {"tags": ",".join(tags)} if tags else {}
+
+
+class PassthroughClient(ProxyClient):
+    # ---- Gemini native passthrough (/gemini/v1beta/...) -----------------
+
+    def gemini_generate(
+        self,
+        key: str,
+        model: str,
+        text: str,
+        *,
+        tools: Optional[Tools] = None,
+        tags: Optional[List[str]] = None,
+    ) -> PassthroughResult:
+        body: Dict[str, object] = {"contents": [{"role": "user", "parts": [{"text": text}]}]}
+        if tools is not None:
+            body["tools"] = tools
+        headers = {
+            "x-goog-api-key": key,
+            "Content-Type": "application/json",
+            **_tag_header(tags),
+        }
+        resp = requests.post(
+            f"{self._base_url}/gemini/v1beta/models/{model}:generateContent",
+            headers=headers,
+            json=body,
+            timeout=self._request_timeout,
+        )
+        return PassthroughResult(
+            resp.status_code, resp.headers.get("x-litellm-call-id"), resp.text
+        )
+
+    def gemini_stream(
+        self, key: str, model: str, text: str, *, tags: Optional[List[str]] = None
+    ) -> PassthroughResult:
+        headers = {
+            "x-goog-api-key": key,
+            "Content-Type": "application/json",
+            **_tag_header(tags),
+        }
+        resp = requests.post(
+            f"{self._base_url}/gemini/v1beta/models/{model}:streamGenerateContent",
+            headers=headers,
+            params={"alt": "sse"},
+            json={"contents": [{"role": "user", "parts": [{"text": text}]}]},
+            stream=True,
+            timeout=self._request_timeout,
+        )
+        call_id = resp.headers.get("x-litellm-call-id")
+        if not (200 <= resp.status_code < 300):
+            return PassthroughResult(resp.status_code, call_id, resp.text)
+        chunks = sum(1 for line in cast("Iterator[bytes]", resp.iter_lines()) if line)
+        return PassthroughResult(resp.status_code, call_id, "<streamed>", chunks)
+
+    # ---- Anthropic native passthrough (/anthropic/v1/messages) ----------
+
+    def anthropic_message(
+        self,
+        key: str,
+        model: str,
+        text: str,
+        *,
+        max_tokens: int = 64,
+        tools: Optional[Tools] = None,
+        stream: bool = False,
+        tags: Optional[List[str]] = None,
+    ) -> PassthroughResult:
+        body: Dict[str, object] = {
+            "model": model,
+            "max_tokens": max_tokens,
+            "messages": [{"role": "user", "content": text}],
+        }
+        if tools is not None:
+            body["tools"] = tools
+        if stream:
+            body["stream"] = True
+        headers = {
+            "x-api-key": key,
+            "anthropic-version": "2023-06-01",
+            "Content-Type": "application/json",
+            **_tag_header(tags),
+        }
+        url = f"{self._base_url}/anthropic/v1/messages"
+        if not stream:
+            resp = requests.post(
+                url, headers=headers, json=body, timeout=self._request_timeout
+            )
+            return PassthroughResult(
+                resp.status_code, resp.headers.get("x-litellm-call-id"), resp.text
+            )
+        resp = requests.post(
+            url, headers=headers, json=body, stream=True, timeout=self._request_timeout
+        )
+        call_id = resp.headers.get("x-litellm-call-id")
+        if not (200 <= resp.status_code < 300):
+            return PassthroughResult(resp.status_code, call_id, resp.text)
+        chunks = sum(1 for line in cast("Iterator[bytes]", resp.iter_lines()) if line)
+        return PassthroughResult(resp.status_code, call_id, "<streamed>", chunks)
+
+
+def build_client() -> PassthroughClient:
+    return PassthroughClient(
+        base_url=PROXY_BASE_URL,
+        master_key=MASTER_KEY,
+        request_timeout=REQUEST_TIMEOUT,
+        poll_timeout=POLL_TIMEOUT,
+        poll_interval=POLL_INTERVAL,
+    )

tests/e2e/llm_translation/test_passthrough_e2e.py

@@ -0,0 +1,158 @@
+"""Live e2e for LLM-translation passthrough endpoints.
+
+Each test sends a NATIVE provider request through the proxy's passthrough route
+and verifies the proxy still logged a costed SpendLogs row
+(call_type="pass_through_endpoint"), correlated by the x-litellm-call-id header.
+
+Covered: gemini ("gemini-2.5-flash") + anthropic ("claude-haiku-4-5"), streaming +
+non-streaming, plus native tool calls. See LLM_TRANSLATION_COVERAGE_MATRIX.md.
+
+A passthrough call returning non-2xx fails hard (never a skip); once it returns
+2xx, a missing or zero-cost SpendLogs row fails too.
+"""
+
+import pytest
+
+from passthrough_client import PassthroughClient, PassthroughResult
+from proxy_client import SpendLogRow, require_successful_call, unique_marker
+
+pytestmark = pytest.mark.e2e
+
+
+def _f(value: object) -> float:
+    return float(value) if isinstance(value, (int, float, str)) else 0.0
+
+
+def _s(value: object) -> str:
+    return str(value) if value is not None else ""
+
+
+def _costed_row(client: PassthroughClient, result: PassthroughResult) -> SpendLogRow:
+    """The passthrough call's logged row, polled until it carries a cost.
+
+    Asserts (not skips) that a 2xx passthrough call produced a costed row - the
+    whole point of passthrough spend tracking.
+    """
+    assert result.call_id, "passthrough response had no x-litellm-call-id header"
+    rows = client.poll_logs_for_request_id(
+        result.call_id,
+        predicate=lambda rs: _f(rs[0].get("spend")) > 0,
+    )
+    assert rows, f"no SpendLogs row for passthrough call_id {result.call_id}"
+    row = rows[0]
+    assert _s(row.get("call_type")) == "pass_through_endpoint"
+    assert _f(row.get("spend")) > 0, f"passthrough call was not costed: {row}"
+    assert _s(row.get("status")) == "success"
+    return row
+
+
+# ---- Gemini passthrough ------------------------------------------------
+
+
+def test_gemini_passthrough_nonstreaming_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    tag = f"e2e-passthrough-{unique_marker()}"
+    result = client.gemini_generate(
+        scoped_key, "gemini-2.5-flash", "Say hello in one word", tags=[tag, "gemini"]
+    )
+    require_successful_call(result)
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "gemini"
+    assert "gemini" in _s(row.get("model"))
+    assert tag in _s(row.get("request_tags")), f"tags not logged: {row.get('request_tags')}"
+
+
+def test_gemini_passthrough_streaming_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    result = client.gemini_stream(scoped_key, "gemini-2.5-flash", "Count to five")
+    require_successful_call(result)
+    assert result.chunks > 0, "streaming passthrough produced no events"
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "gemini"
+
+
+def test_gemini_passthrough_tool_call_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    result = client.gemini_generate(
+        scoped_key,
+        "gemini-2.5-flash",
+        "What is the weather in Paris? Use the get_weather tool.",
+        tools=[
+            {
+                "functionDeclarations": [
+                    {
+                        "name": "get_weather",
+                        "description": "Get the weather for a city",
+                        "parameters": {
+                            "type": "object",
+                            "properties": {"city": {"type": "string"}},
+                            "required": ["city"],
+                        },
+                    }
+                ]
+            }
+        ],
+    )
+    require_successful_call(result)
+    assert "functionCall" in result.body, "gemini did not emit a tool call"
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "gemini"
+
+
+# ---- Anthropic passthrough ---------------------------------------------
+
+
+def test_anthropic_passthrough_nonstreaming_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    result = client.anthropic_message(scoped_key, "claude-haiku-4-5", "Say hello")
+    require_successful_call(result)
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "anthropic"
+    assert "claude" in _s(row.get("model"))
+
+
+def test_anthropic_passthrough_streaming_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    result = client.anthropic_message(
+        scoped_key, "claude-haiku-4-5", "Count to five", stream=True
+    )
+    require_successful_call(result)
+    assert result.chunks > 0, "streaming passthrough produced no events"
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "anthropic"
+
+
+def test_anthropic_passthrough_tool_call_logs_cost(
+    client: PassthroughClient, scoped_key: str
+) -> None:
+    result = client.anthropic_message(
+        scoped_key,
+        "claude-haiku-4-5",
+        "What is the weather in Paris? Use the get_weather tool.",
+        tools=[
+            {
+                "name": "get_weather",
+                "description": "Get the weather for a city",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {"city": {"type": "string"}},
+                    "required": ["city"],
+                },
+            }
+        ],
+    )
+    require_successful_call(result)
+    assert "tool_use" in result.body, "anthropic did not emit a tool call"
+
+    row = _costed_row(client, result)
+    assert _s(row.get("custom_llm_provider")) == "anthropic"

tests/pyrightconfig.json

@@ -0,0 +1,11 @@
+{
+    "include": ["e2e"],
+    "exclude": ["**/node_modules", "**/__pycache__"],
+    "pythonVersion": "3.12",
+    "typeCheckingMode": "strict",
+    "enableTypeIgnoreComments": false,
+    "reportMissingImports": false,
+    "reportPrivateImportUsage": false,
+    "reportExplicitAny": "error",
+    "reportAny": "error"
+}