From 7915b0fd6e7b465a5f7bb590d12414a39bfbd928 Mon Sep 17 00:00:00 2001 From: Nick Nikolakakis Date: Sun, 16 Aug 2026 08:30:00 +0300 Subject: [PATCH 1/2] Answer real requests in the local e2e mock engine The mock engine returned one canned response to any POST and ignored the request body. That caps what the e2e can check at a status code: a reply that never reached the engine, or one a proxy invented, looks the same as a real one, and there is nothing to assert about streaming or bad input because the mock has no such behaviour. It now echoes the last user turn and the requested model, so a caller can tell its body arrived intact; serves an SSE stream of chat.completion.chunk frames terminated by [DONE] when a request sets stream; and rejects an unparseable body or one with no messages with a 400, and an unknown path with a 404, instead of answering 200 either way. It serves HTTP/1.1 so the stream is chunked the way an engine's is, which means keep-alive, so it also moves to ThreadingHTTPServer: a single-threaded server would hold the probes and the endpoint-picker behind one client. Reading the body also means reading it the way it arrives. The endpoint picker is an ext_proc filter, and Envoy strips content-length from a request it hands to one unless allow_content_length_header is set, which it is not; the body then reaches the engine chunked. A read that trusts content-length sees zero bytes and answers 400 to everything, so the mock decodes a chunked body as well. The cost is size. The ModelDeployment container template is a curated subset of name, image, command, args and env, so there is no volume to load a file from and the server stays inline in args, now about twice as long. Loading it from a ConfigMap through env.valueFrom is possible if it grows further. Towards #368. Signed-off-by: Nick Nikolakakis --- e2e/manifests/40-model-deployment.yaml | 66 ++++++++++++++++++++++---- 1 file changed, 57 insertions(+), 9 deletions(-) diff --git a/e2e/manifests/40-model-deployment.yaml b/e2e/manifests/40-model-deployment.yaml index f0d76ef14..db90c1114 100644 --- a/e2e/manifests/40-model-deployment.yaml +++ b/e2e/manifests/40-model-deployment.yaml @@ -37,6 +37,8 @@ spec: - | import json, http.server class H(http.server.BaseHTTPRequestHandler): + # HTTP/1.1 so the streaming path can chunk, the way a real engine does. + protocol_version = "HTTP/1.1" def _s(self, o, c=200): b = json.dumps(o).encode() self.send_response(c) @@ -44,6 +46,33 @@ spec: self.send_header("content-length", str(len(b))) self.end_headers() self.wfile.write(b) + def _sse(self, frames): + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.send_header("cache-control", "no-cache") + self.send_header("transfer-encoding", "chunked") + self.end_headers() + for f in [*frames, "[DONE]"]: + b = ("data: " + (f if isinstance(f, str) else json.dumps(f)) + "\n\n").encode() + self.wfile.write(b"%X\r\n" % len(b) + b + b"\r\n") + self.wfile.flush() + self.wfile.write(b"0\r\n\r\n") + def _body(self): + # The endpoint picker's ext_proc drops content-length, so bodies arrive chunked. + if "chunked" not in (self.headers.get("transfer-encoding") or "").lower(): + return self.rfile.read(int(self.headers.get("content-length") or 0)) + b = b"" + while n := int(self.rfile.readline().split(b";")[0] or b"0", 16): + b += self.rfile.read(n) + self.rfile.readline() + self.rfile.readline() + return b + def _read(self): + try: + r = json.loads(self._body() or b"{}") + except ValueError: + return None + return r if isinstance(r, dict) and r.get("messages") else None def do_GET(self): if self.path == "/health": self._s({"status": "ok"}) @@ -52,27 +81,46 @@ spec: else: self._s({"error": "not found"}, 404) def do_POST(self): - n = int(self.headers.get("content-length") or 0) - self.rfile.read(n) - text = "Hello from the Modelplane local mock engine." - if self.path.startswith("/v1/messages"): + req = self._read() + anthropic = self.path.startswith("/v1/messages") + if not anthropic and not self.path.startswith("/v1/chat/completions"): + self._s({"error": "not found"}, 404) + elif req is None: + self._s({"error": "messages is required"}, 400) + else: + self._answer(req, anthropic) + def _answer(self, req, anthropic): + model = req.get("model") or "mock" + # Echo the last user turn back, so a caller can tell its request reached the engine. + turns = [m.get("content") or "" for m in req["messages"] if m.get("role") == "user"] + text = "Echo: " + (turns[-1] if turns else "") + if anthropic: # Anthropic Messages API; vLLM serves it alongside the OpenAI routes. self._s({ "id": "msg-mock", "type": "message", "role": "assistant", - "model": "mock", "stop_reason": "end_turn", + "model": model, "stop_reason": "end_turn", "content": [{"type": "text", "text": text}], - "usage": {"input_tokens": 1, "output_tokens": 1}, + "usage": {"input_tokens": len(turns), "output_tokens": 1}, }) + elif req.get("stream"): + head = {"id": "chatcmpl-mock", "object": "chat.completion.chunk", "model": model} + frames = [dict(head, choices=[{"index": 0, "delta": {"role": "assistant"}}])] + frames += [dict(head, choices=[{"index": 0, "delta": {"content": w + " "}}]) for w in text.split()] + frames.append(dict(head, choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}])) + self._sse(frames) else: self._s({ - "id": "chatcmpl-mock", "object": "chat.completion", "model": "mock", + "id": "chatcmpl-mock", "object": "chat.completion", "model": model, "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}], - "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + "usage": {"prompt_tokens": len(turns), "completion_tokens": 1, + "total_tokens": len(turns) + 1}, }) def log_message(self, *a): pass - http.server.HTTPServer(("0.0.0.0", 8000), H).serve_forever() + # Threaded: HTTP/1.1 keeps connections alive, so a single-threaded server + # would block the probes and the endpoint-picker behind one client. + http.server.ThreadingHTTPServer(("0.0.0.0", 8000), H).serve_forever() # The ModelDeployment container template is a curated subset # (name/image/command/args/env); the composition wires the port, # readiness, and any GPU resources itself (it assumes port 8000). So From 4da464a44a2740ecbf53845ee5a3c4181054faae Mon Sep 17 00:00:00 2001 From: Nick Nikolakakis Date: Sun, 16 Aug 2026 08:30:15 +0300 Subject: [PATCH 2/2] Assert what the served endpoints return, not just a 200 The local e2e curls /v1/chat/completions and /v1/messages and checks the status code. That proves the path is wired, and nothing else: a regression that answers 200 with a wrong shape, drops the request body on the way to the engine, or buffers a stream into one blob passes unnoticed. This replaces the two curl pods with a stdlib unittest suite under e2e/verify, pointed at the ModelService through MODELPLANE_ADDRESS. It asserts the OpenAI and Anthropic response shapes, that a marker in the last user turn comes back from the engine, that a streamed request returns event-stream frames whose content deltas reassemble into the message and end in [DONE], that a malformed body and a body with no messages are refused, and that /v1/models lists the served model. It waits for the route to serve before the first assertion, since the address publishes before the path carries traffic. The address is on the kind Docker subnet, so the suite runs from a pod on the control plane like the curls did. It goes in as a ConfigMap and runs under the same python image the mock engine uses: urllib and unittest only, so there is nothing to build and no install step. run.sh polls the pod's phase rather than using kubectl wait, which has no single condition for "ran, either way", prints the logs so a red run says which assertion failed, and exits non-zero unless the pod succeeded. The ruff and license checks grow a path to cover it, since both were scoped to functions/ and the docs validator and would otherwise leave new Python unchecked. Fixes #368. Signed-off-by: Nick Nikolakakis --- e2e/README.md | 16 ++-- e2e/run.sh | 108 ++++++++++++----------- e2e/verify/test_serving.py | 172 +++++++++++++++++++++++++++++++++++++ nix/apps.nix | 4 +- nix/checks.nix | 17 ++-- 5 files changed, 253 insertions(+), 64 deletions(-) create mode 100644 e2e/verify/test_serving.py diff --git a/e2e/README.md b/e2e/README.md index 3e257d8cf..1a7b1e92c 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -94,7 +94,7 @@ against. The control-plane cluster needs no DRA. ```bash nix run .#e2e # bring up both clusters + deploy the mock model -nix run .#e2e -- --verify # same, then wait for readiness and assert a live 200 +nix run .#e2e -- --verify # same, then wait for readiness and run the behavioral suite nix run .#e2e -- --clean # tear both clusters down ``` @@ -119,10 +119,16 @@ kubectl run curl -n ml-team --rm -it --image=curlimages/curl@sha256:7c12af72ceb3 -d '{"model":"mock","max_tokens":16,"messages":[{"role":"user","content":"hi"}]}' ``` -`--verify` runs both of those (OpenAI then Anthropic) and exits non-zero on -failure. It's the exact command the `E2E` CI workflow runs, so a green `--verify` -locally and a green CI run mean the same thing; use the manual curls above to -poke the endpoints interactively. +`--verify` goes further than those curls: it runs [`verify/test_serving.py`](verify/test_serving.py) +from a pod on the control plane and asserts what came back — the response shapes +both surfaces promise, that the last user turn reaches the engine, that a +streamed response arrives as `text/event-stream` frames rather than one buffered +blob, and that malformed input and unrouted paths fail. The suite goes in as a +ConfigMap and runs under a stock `python:3.12-alpine` pod; it imports only the +stdlib, so nothing is built or installed. It exits non-zero on failure and is the +exact command the `E2E` CI workflow runs, so a green `--verify` locally and a +green CI run mean the same thing; use the manual curls above to poke the +endpoints interactively. ## How it's structured diff --git a/e2e/run.sh b/e2e/run.sh index 9534f2ce4..982a1de16 100644 --- a/e2e/run.sh +++ b/e2e/run.sh @@ -20,9 +20,9 @@ WL=modelplane-e2e-workload # v1.34.2 or newer: older kubelets deadlock on an idle DRA connection (k/k#133934). WL_NODE_IMAGE=kindest/node:v1.34.8@sha256:02722c2dedddcfc00febf5d27fbeb9b7b2c14294c82109ff4a85d89ac9ba3256 METALLB_URL=https://raw.githubusercontent.com/metallb/metallb/v0.14.8/config/manifests/metallb-native.yaml -# Pinned by digest (a multi-arch manifest list) so a moving :latest can't flake -# the verify curl pod. -CURL_IMAGE=curlimages/curl@sha256:7c12af72ceb38b7432ab85e1a265cff6ae58e06f95539d539b654f2cfa64bb13 +# The same python:3.12-alpine digest the mock engine runs, so the e2e pins one +# image. The suite is stdlib-only, so the stock image needs no pip install. +PYTHON_IMAGE=python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df ROOT="$(git rev-parse --show-toplevel)" log() { printf '\n\033[1;34m==> %s\033[0m\n' "$*"; } @@ -186,9 +186,9 @@ fi # --verify: project run returns once the config is healthy and the resources are # applied, so the serving-stack install and model rollout are still reconciling. -# Wait for the ModelService to publish an address, then route a real request to -# the engine and assert a 200. Any failure exits non-zero — that is what makes -# this usable as a CI gate. +# Wait for the ModelService to publish an address, then run the behavioral suite +# in e2e/verify against it. Any failure exits non-zero — that is what makes this +# usable as a CI gate. log "Verifying the model serves end to end" ns=ml-team svc=mock @@ -206,52 +206,62 @@ done log "ModelService address: $addr" # The address is on the kind Docker subnet the host can't route to on macOS, so -# curl from a pod on the control plane, reading the status from the pod's logs -# (not `run -i`, whose attach drops output on a headless runner). curl_status -# runs one throwaway pod per call and echoes the HTTP code; it polls the logs -# (curl writes the code once, then exits) so a failed attempt costs seconds, and -# a unique pod name per call keeps retries from reading a prior pod's output. -curl_status() { - local pod="$1" url="$2" - shift 2 - kubectl --context "$cpctx" -n "$ns" run "$pod" --restart=Never \ - --labels=app.kubernetes.io/name=e2e-verify --image="$CURL_IMAGE" \ - --command -- curl -sS --max-time 15 -o /dev/null -w '%{http_code}' "$url" "$@" \ - >/dev/null 2>&1 || true - local c="" - for _ in $(seq 1 30); do - c="$(kubectl --context "$cpctx" -n "$ns" logs "$pod" 2>/dev/null | tr -dc '0-9' || true)" - [ -n "$c" ] && break - sleep 2 - done - printf '%s' "$c" -} +# the suite runs from a pod on the control plane. It ships as a ConfigMap rather +# than an image we'd have to build and push, and the stock python image runs it +# as-is because it imports nothing outside the stdlib. +log "Running the behavioral suite (e2e/verify) against $addr" +kubectl --context "$cpctx" -n "$ns" create configmap e2e-verify-suite \ + --from-file=test_serving.py="$ROOT/e2e/verify/test_serving.py" \ + --dry-run=client -o yaml | kubectl --context "$cpctx" apply -f - + +# A rerun against a live cluster would otherwise hit an immutable completed pod. +kubectl --context "$cpctx" -n "$ns" delete pod e2e-verify --now >/dev/null 2>&1 || true +kubectl --context "$cpctx" apply -f - </dev/null || true)" + case "$phase" in + Succeeded | Failed) break ;; + esac sleep 10 done -[ "$code" = "200" ] || { - echo "verify: $addr/v1/chat/completions did not return 200 within retries (last: ${code:-none})" >&2 - kubectl --context "$cpctx" -n "$ns" delete pod -l app.kubernetes.io/name=e2e-verify --now >/dev/null 2>&1 || true - exit 1 -} -# Anthropic Messages API on the same address: vLLM serves /v1/messages alongside -# the OpenAI routes (PR #360) and the route preserves the path. Serving is up by -# now, so one attempt suffices. -ant='{"model":"'"$svc"'","max_tokens":16,"messages":[{"role":"user","content":"ping"}]}' -mcode="$(curl_status e2e-verify-anthropic "$addr/v1/messages" -H 'content-type: application/json' -H 'anthropic-version: 2023-06-01' -d "$ant")" -log "verify (Anthropic /v1/messages): HTTP ${mcode:-none}" -kubectl --context "$cpctx" -n "$ns" delete pod -l app.kubernetes.io/name=e2e-verify --now >/dev/null 2>&1 || true -[ "$mcode" = "200" ] || { - echo "verify: $addr/v1/messages did not return 200 (last: ${mcode:-none})" >&2 +# unittest writes to stderr, and the logs are the failure report on a red run. +kubectl --context "$cpctx" -n "$ns" logs e2e-verify 2>&1 || true +kubectl --context "$cpctx" -n "$ns" delete pod e2e-verify --now >/dev/null 2>&1 || true +kubectl --context "$cpctx" -n "$ns" delete configmap e2e-verify-suite >/dev/null 2>&1 || true +[ "$phase" = "Succeeded" ] || { + echo "verify: the behavioral suite did not pass (pod phase: ${phase:-none})" >&2 exit 1 } -log "End to end OK: $addr serves OpenAI (/v1/chat/completions) and Anthropic (/v1/messages)" +log "End to end OK: $addr passes the behavioral suite in e2e/verify" diff --git a/e2e/verify/test_serving.py b/e2e/verify/test_serving.py new file mode 100644 index 000000000..17e97b73d --- /dev/null +++ b/e2e/verify/test_serving.py @@ -0,0 +1,172 @@ +# 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. + +"""Behavioral checks against a running ModelService. + +A 200 only proves the path is wired. These assert what came back: the response +shapes the OpenAI and Anthropic surfaces promise, that a request body reaches +the engine intact, that a streamed response survives the two-hop route as +event-stream frames, and that bad input fails instead of returning a 200 with +an error in the body. + +MODELPLANE_ADDRESS is the ModelService's status.address. It's on the kind +Docker subnet, so this runs from a pod on the control plane (see run.sh), not +from the host. +""" + +import email.message +import json +import os +import time +import unittest +import urllib.error +import urllib.request + +ADDRESS = os.environ.get("MODELPLANE_ADDRESS", "").rstrip("/") +MODEL = os.environ.get("MODELPLANE_MODEL", "mock") +READY_TIMEOUT = float(os.environ.get("MODELPLANE_READY_TIMEOUT", "300")) + +OK = 200 +BAD_REQUEST = 400 +ANTHROPIC_HEADERS = {"anthropic-version": "2023-06-01"} + +Response = tuple[int, dict[str, str], bytes] + + +def received(message: email.message.Message) -> dict[str, str]: + """Header names are case-insensitive, and a proxy is free to re-case what the engine sent.""" + return {name.lower(): value for name, value in message.items()} + + +def call( + path: str, data: bytes | None = None, *, headers: dict[str, str] | None = None, timeout: float = 30.0 +) -> Response: + """POST data (GET when it's None) and return the status, headers and body, 4xx included.""" + req = urllib.request.Request(f"{ADDRESS}{path}", data=data, method="GET" if data is None else "POST") + req.add_header("content-type", "application/json") + for name, value in (headers or {}).items(): + req.add_header(name, value) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, received(resp.headers), resp.read() + except urllib.error.HTTPError as err: + with err: + return err.code, received(err.headers), err.read() + + +def chat(*turns: str, stream: bool = False) -> bytes: + """An OpenAI chat request. Turns alternate user, assistant, user, so the last one is the user's.""" + messages = [{"role": "user" if i % 2 == 0 else "assistant", "content": t} for i, t in enumerate(turns)] + payload: dict[str, object] = {"model": MODEL, "messages": messages} + if stream: + payload["stream"] = True + return json.dumps(payload).encode() + + +def frames(body: bytes) -> list[str]: + return [line[len("data: ") :] for line in body.decode().splitlines() if line.startswith("data: ")] + + +def setUpModule() -> None: + """Wait for the route to serve. The address publishes before the path carries traffic.""" + if not ADDRESS: + raise RuntimeError("MODELPLANE_ADDRESS is not set") + deadline = time.monotonic() + READY_TIMEOUT + status = 0 + while True: + try: + status, _, _ = call("/v1/chat/completions", chat("ready?"), timeout=15.0) + except OSError: + status = 0 + if status == OK: + return + if time.monotonic() >= deadline: + raise AssertionError( + f"{ADDRESS} did not serve a chat completion within {READY_TIMEOUT:.0f}s (last status {status})" + ) + time.sleep(5) + + +class ChatCompletions(unittest.TestCase): + def test_response_shape(self) -> None: + status, _, body = call("/v1/chat/completions", chat("ping")) + self.assertEqual(status, OK) + got = json.loads(body) + self.assertEqual(got["object"], "chat.completion") + self.assertEqual(got["model"], MODEL) + self.assertEqual(len(got["choices"]), 1) + choice = got["choices"][0] + self.assertEqual(choice["message"]["role"], "assistant") + self.assertTrue(choice["message"]["content"], "empty assistant content") + self.assertEqual(choice["finish_reason"], "stop") + self.assertIn("usage", got) + + def test_last_user_turn_reaches_the_engine(self) -> None: + marker = "marker-4d91" + status, _, body = call("/v1/chat/completions", chat("first", "an earlier answer", marker)) + self.assertEqual(status, OK) + content = json.loads(body)["choices"][0]["message"]["content"] + self.assertIn(marker, content, "the engine did not see the last user turn") + + def test_streaming_returns_event_stream_frames(self) -> None: + status, headers, body = call("/v1/chat/completions", chat("one two", stream=True)) + self.assertEqual(status, OK) + self.assertIn("text/event-stream", headers.get("content-type", ""), f"headers: {headers}") + got = frames(body) + self.assertGreater(len(got), 1, "a stream collapsed to a single frame") + self.assertEqual(got[-1], "[DONE]") + self.assertEqual(json.loads(got[0])["object"], "chat.completion.chunk") + streamed = "".join(json.loads(f)["choices"][0]["delta"].get("content", "") for f in got[:-1]) + self.assertIn("one two", streamed) + + def test_malformed_body_is_rejected(self) -> None: + status, _, _ = call("/v1/chat/completions", b"not json") + self.assertGreaterEqual(status, BAD_REQUEST, "malformed JSON was accepted") + + def test_request_without_messages_is_rejected(self) -> None: + status, _, _ = call("/v1/chat/completions", json.dumps({"model": MODEL}).encode()) + self.assertGreaterEqual(status, BAD_REQUEST, "a request with no messages was accepted") + + +class Messages(unittest.TestCase): + """The Anthropic surface, served on the same address.""" + + def test_response_shape(self) -> None: + payload = json.dumps({"model": MODEL, "max_tokens": 16, "messages": [{"role": "user", "content": "ping"}]}) + status, _, body = call("/v1/messages", payload.encode(), headers=ANTHROPIC_HEADERS) + self.assertEqual(status, OK) + got = json.loads(body) + self.assertEqual(got["type"], "message") + self.assertEqual(got["role"], "assistant") + self.assertEqual(got["model"], MODEL) + self.assertEqual(got["content"][0]["type"], "text") + self.assertTrue(got["content"][0]["text"], "empty text block") + self.assertEqual(got["stop_reason"], "end_turn") + self.assertIn("input_tokens", got["usage"]) + self.assertIn("output_tokens", got["usage"]) + + +class Routing(unittest.TestCase): + def test_models_lists_the_served_model(self) -> None: + status, _, body = call("/v1/models") + self.assertEqual(status, OK) + self.assertIn(MODEL, [m["id"] for m in json.loads(body)["data"]]) + + def test_unknown_path_does_not_serve(self) -> None: + status, _, _ = call("/v1/nonexistent", chat("ping")) + self.assertGreaterEqual(status, BAD_REQUEST, "an unrouted path returned a success") + + +if __name__ == "__main__": + unittest.main() diff --git a/nix/apps.nix b/nix/apps.nix index a2723658c..94180dbdf 100644 --- a/nix/apps.nix +++ b/nix/apps.nix @@ -46,8 +46,8 @@ find . -name '*.sh' -type f -exec shellcheck {} + echo "Formatting and linting Python..." - ruff format functions/ - ruff check --fix functions/ + ruff format functions/ e2e/verify/ + ruff check --fix functions/ e2e/verify/ echo "Refreshing uv.lock..." uv lock diff --git a/nix/checks.nix b/nix/checks.nix index bd5f7e286..b85f7627c 100644 --- a/nix/checks.nix +++ b/nix/checks.nix @@ -106,8 +106,8 @@ in cp -r ${self} src chmod -R u+w src cd src - ruff format --check functions/ docs/utils/validate/ - ruff check functions/ docs/utils/validate/ + ruff format --check functions/ docs/utils/validate/ e2e/verify/ + ruff check functions/ docs/utils/validate/ e2e/verify/ mkdir -p $out touch $out/.python-checks-passed ''; @@ -141,11 +141,12 @@ in ''; # Fail if any hand-written source file is missing its Apache 2.0 license - # header. Scoped to the files we author: the composition functions and the - # docs manifest validator. Generated models under schemas/python carry their - # own codegen banner, and config (*.toml) and vendored upstream CRDs (*.yaml) - # are excluded. addlicense -check only reads, so it runs against the store - # path directly. Run 'nix run .#fix' to add any missing headers. + # header. Scoped to the files we author: the composition functions, the docs + # manifest validator, and the e2e verify suite. Generated models under + # schemas/python carry their own codegen banner, and config (*.toml) and + # vendored upstream CRDs (*.yaml) are excluded. addlicense -check only reads, + # so it runs against the store path directly. Run 'nix run .#fix' to add any + # missing headers. license = pkgs.runCommand "modelplane-license-check" { @@ -157,7 +158,7 @@ in -ignore '**/*.toml' \ -ignore '**/*.yaml' \ -ignore '**/*.yml' \ - functions/ docs/utils/validate/ nix.sh + functions/ docs/utils/validate/ e2e/verify/ nix.sh mkdir -p $out touch $out/.license-check-passed '';