From 8d2053e588e61cd2d9f89b3a5ce3d38bd8ec8a98 Mon Sep 17 00:00:00 2001 From: Nick Nikolakakis Date: Tue, 13 Oct 2026 19:12:08 +0300 Subject: [PATCH 1/2] Make the e2e mock engine echo, stream and reject bad requests The mock engine answered every chat completion with the same fixed text, never streamed, and treated a body that wasn't JSON as an empty one. So a test could only check that a response came back, not that the caller's request reached the engine intact, that a streamed response survived the gateway as event-stream frames, or that a bad request failed instead of returning a 200. The mock now echoes the last user turn, streams chunk frames ending in [DONE] when the request sets stream, and answers 400 to a body that isn't JSON or carries no messages, as vLLM does. Usage stays fixed, so the metering test's expected record is unchanged. Towards #368. Signed-off-by: Nick Nikolakakis --- e2e/manifests/40-model-deployment.yaml | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/e2e/manifests/40-model-deployment.yaml b/e2e/manifests/40-model-deployment.yaml index e32a3508c..01a724f16 100644 --- a/e2e/manifests/40-model-deployment.yaml +++ b/e2e/manifests/40-model-deployment.yaml @@ -93,7 +93,8 @@ spec: try: body = json.loads(raw or b"{}") except Exception: - body = {} + self._s({"error": {"message": "body is not JSON", "type": "BadRequestError"}}, 400) + return # A real engine only answers to the name it was started # with. Rejecting anything else is what makes the # InferenceGateway's model rewriting observable: without it the @@ -103,7 +104,12 @@ spec: % (body.get("model"), served), "type": "NotFoundError"}}, 404) return - text = "Hello from the Modelplane local mock engine." + if not body.get("messages"): + self._s({"error": {"message": "messages is required", "type": "BadRequestError"}}, 400) + return + # Echo the last user turn, so a caller can tell its request reached the engine intact. + turns = [m.get("content") or "" for m in body["messages"] if m.get("role") == "user"] + text = "Echo: " + (turns[-1] if turns else "") if self.path.startswith("/v1/messages"): # Anthropic Messages API; vLLM serves it alongside the OpenAI routes. self._s({ @@ -112,6 +118,17 @@ spec: "content": [{"type": "text", "text": text}], "usage": {"input_tokens": 12, "output_tokens": 9}, }) + elif body.get("stream"): + head = {"id": "chatcmpl-mock", "object": "chat.completion.chunk", "model": served} + deltas = [{"role": "assistant"}] + [{"content": w + " "} for w in text.split()] + frames = [dict(head, choices=[{"index": 0, "delta": d}]) for d in deltas] + frames.append(dict(head, choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}])) + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + for f in frames: + self.wfile.write(b"data: " + json.dumps(f).encode() + b"\n\n") + self.wfile.write(b"data: [DONE]\n\n") else: self._s({ "id": "chatcmpl-mock", "object": "chat.completion", "model": served, From 22c9ac1c390ae242cf7948745475a32ba0ddb205 Mon Sep 17 00:00:00 2001 From: Nick Nikolakakis Date: Tue, 13 Oct 2026 19:12:09 +0300 Subject: [PATCH 2/2] Test that requests and responses cross the InferenceGateway intact The e2e tests check that the InferenceGateway routes a request and rewrites its model, but not what comes back. A gateway that mangled the body, buffered a stream into one response, or turned a bad request into a 200 would pass them. test_serving.py checks the OpenAI and Anthropic response shapes, that the engine sees the caller's last user turn, that a streamed completion arrives as event-stream frames, and that a malformed body, a request without messages, and an unrouted path all fail. Client.request gains raw, to send a body that isn't JSON, and Response gains content_type, which the streaming test needs. Fixes #368. Signed-off-by: Nick Nikolakakis --- e2e/README.md | 1 + e2e/gateway.py | 21 ++++--- e2e/tests/test_serving.py | 128 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 143 insertions(+), 7 deletions(-) create mode 100644 e2e/tests/test_serving.py diff --git a/e2e/README.md b/e2e/README.md index f5978ded4..9df024469 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -191,6 +191,7 @@ e2e/ conftest.py # fixtures: the clusters, curl pods, readiness test_auth.py # the InferenceGateway refuses unknown callers test_routing.py # it routes OpenAI and Anthropic requests + test_serving.py # requests and responses cross it intact, bad ones fail test_metering.py # it logs a usage record per request test_cluster_gateway.py # the cluster gateway requires a client certificate test_telemetry.py # the engine's series reach the collector's sink diff --git a/e2e/gateway.py b/e2e/gateway.py index dce5fe2d1..32a7cf591 100644 --- a/e2e/gateway.py +++ b/e2e/gateway.py @@ -54,6 +54,7 @@ class Response: status: int # The body, or why no HTTP response came back. body: str + content_type: str = "" def json(self) -> typing.Any: # noqa: ANN401 - a JSON body can decode to any type. """Decode the body as JSON.""" @@ -68,20 +69,26 @@ class Client: namespace: str pod: str - def request(self, url: str, headers: dict[str, str], body: object | None = None) -> Response: - """Send a request, a POST of body as JSON if there is one and otherwise a GET.""" - curl = ["curl", "--silent", "--show-error", "--max-time", "15", "--write-out", "\n%{http_code}", url] + def request( + self, url: str, headers: dict[str, str], body: object | None = None, *, raw: str | None = None + ) -> Response: + """Send a request, a POST of body as JSON (or of raw as is) if there is one and otherwise a GET.""" + write_out = "\n%{content_type}\n%{http_code}" + curl = ["curl", "--silent", "--show-error", "--max-time", "15", "--write-out", write_out, url] for name, value in headers.items(): curl += ["--header", f"{name}: {value}"] if body is not None: - curl += ["--header", "content-type: application/json", "--data", json.dumps(body)] + raw = json.dumps(body) + if raw is not None: + curl += ["--header", "content-type: application/json", "--data", raw] result = self.cluster.exec(self.pod, self.namespace, curl) # curl exits non-zero only when no HTTP response came back. if result.code != 0: return Response(status=0, body=result.stderr.strip()) - # --write-out puts the status on a line of its own, after the body. - text, _, status = result.stdout.rpartition("\n") - return Response(status=int(status), body=text) + # --write-out puts the content type and status on lines of their own, after the body. + rest, _, status = result.stdout.rpartition("\n") + text, _, content_type = rest.rpartition("\n") + return Response(status=int(status), body=text, content_type=content_type) def connect(self, url: str) -> int: """GET a URL without verifying the server's certificate, and return curl's exit code.""" diff --git a/e2e/tests/test_serving.py b/e2e/tests/test_serving.py new file mode 100644 index 000000000..4eb792941 --- /dev/null +++ b/e2e/tests/test_serving.py @@ -0,0 +1,128 @@ +# Copyright 2026 The Modelplane Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests that requests and responses cross the InferenceGateway intact.""" + +import collections.abc +import dataclasses +import json + +import pytest + +from e2e import gateway + + +def test_chat_completion(serving: gateway.Serving, control_plane_client: gateway.Client) -> None: + """An OpenAI chat completion comes back in the shape the OpenAI API promises.""" + r = control_plane_client.request( + f"{serving.openai}/chat/completions", + gateway.BEARER, + {"model": serving.model, "messages": [{"role": "user", "content": "ping"}]}, + ) + assert r.status == 200 + got = r.json() + assert got["object"] == "chat.completion" + assert len(got["choices"]) == 1 + assert got["choices"][0]["message"]["role"] == "assistant" + assert got["choices"][0]["message"]["content"] + assert got["choices"][0]["finish_reason"] == "stop" + assert "usage" in got + + +def test_message(serving: gateway.Serving, control_plane_client: gateway.Client) -> None: + """An Anthropic message comes back in the shape the Messages API promises.""" + r = control_plane_client.request( + f"{serving.anthropic}/messages", + {"x-api-key": gateway.CALLER_KEY, "anthropic-version": "2023-06-01"}, + {"model": serving.model, "max_tokens": 16, "messages": [{"role": "user", "content": "ping"}]}, + ) + assert r.status == 200 + got = r.json() + assert got["type"] == "message" + assert got["role"] == "assistant" + assert got["content"][0]["type"] == "text" + assert got["content"][0]["text"] + assert got["stop_reason"] == "end_turn" + assert {"input_tokens", "output_tokens"} <= got["usage"].keys() + + +def test_last_user_turn(serving: gateway.Serving, control_plane_client: gateway.Client) -> None: + """The engine sees the caller's last user turn, so the gateway forwards the body intact.""" + messages = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "an earlier answer"}, + {"role": "user", "content": "marker-4d91"}, + ] + r = control_plane_client.request( + f"{serving.openai}/chat/completions", gateway.BEARER, {"model": serving.model, "messages": messages} + ) + assert r.status == 200 + assert "marker-4d91" in r.json()["choices"][0]["message"]["content"] + + +def test_streaming(serving: gateway.Serving, control_plane_client: gateway.Client) -> None: + """A streamed chat completion comes back as event-stream frames, not one buffered body.""" + r = control_plane_client.request( + f"{serving.openai}/chat/completions", + gateway.BEARER, + {"model": serving.model, "messages": [{"role": "user", "content": "one two"}], "stream": True}, + ) + assert r.status == 200 + assert r.content_type.startswith("text/event-stream"), r + frames = [line.removeprefix("data: ") for line in r.body.splitlines() if line.startswith("data: ")] + assert len(frames) > 1, r + assert frames[-1] == "[DONE]" + chunks = [json.loads(f) for f in frames[:-1]] + assert {c["object"] for c in chunks} == {"chat.completion.chunk"} + assert "one two" in "".join(c["choices"][0]["delta"].get("content", "") for c in chunks) + + +@dataclasses.dataclass +class Case: + """A request the InferenceGateway fails rather than serves.""" + + name: str + reason: str + path: str + # The request body, given the model the caller names. + raw: collections.abc.Callable[[str], str] + + +REJECTED_CASES = [ + Case( + name="MalformedBody", + reason="A body that isn't JSON fails, rather than returning a 200.", + path="/chat/completions", + raw=lambda _: "not json", + ), + Case( + name="NoMessages", + reason="A chat completion without messages fails, rather than returning a 200.", + path="/chat/completions", + raw=lambda model: json.dumps({"model": model}), + ), + Case( + name="UnknownPath", + reason="A path no route serves fails, rather than returning a 200.", + path="/nonexistent", + raw=lambda model: json.dumps({"model": model, "messages": [{"role": "user", "content": "ping"}]}), + ), +] + + +@pytest.mark.parametrize("case", REJECTED_CASES, ids=lambda case: case.name) +def test_rejected(case: Case, serving: gateway.Serving, control_plane_client: gateway.Client) -> None: + """The InferenceGateway answers a bad request with an error status.""" + r = control_plane_client.request(f"{serving.openai}{case.path}", gateway.BEARER, raw=case.raw(serving.model)) + assert r.status >= 400, case.reason