fix(integration): green gate with tracked streaming skip, gemini coverage, robust probe

Addresses coordinator review of the Task 15 integration suite: 1 critical
ruling, 5 important, 3 folded. make integration now exits 0.

- integration_test.go: TestStreamingIncrementalityProbe restructured —
  content assertion (5 deltas + [DONE], in order) always runs; the raw-TCP
  timing assertion (unchanged, >=3 gaps >=100ms) is gated behind
  INTEGRATION_STREAMING=1 with t.Skipf naming traefik/traefik#10269,
  verified to still genuinely fail when re-armed (not silently disabled).
- Makefile: integration-up/integration-wait moved from prerequisites into
  the integration recipe body as go test ./... -count=1
ok  	github.com/lukaszraczylo/traefik-llmgateway	6.065s sub-invocations with per-stage
  status capture, so integration-down always runs even when bring-up or
  the health wait fails; verified with a synthetic failure. Added
  .NOTPARALLEL (the integration targets are stateful).
- translate_gemini.go: fixed the same yaegi multi-value-assignment panic
  shape as the prior sse.go fix, on the live max_tokens path. Added a
  gemini chat completion assertion to integration_test.go (renamed
  TestUnifiedChatAllProviders) so this path now runs under real Yaegi.
- providers.go, translate_anthropic.go: documented the never-wrap
  contract on providerHTTPError/translateError (routes_unified.go asserts
  their concrete type directly; errors.As panics under yaegi on these).
- integration_test.go: TestCrossReplicaLimits now FLUSHALLs Redis first,
  for idempotence under integration-keep.
- mock/main.go: probe's header-terminator search now accumulates a
  rolling buffer across reads instead of only checking the latest one.
- Folded: hot-reload deadline 10s -> 20s; dynamic.yml's real-upstream
  baseUrl is now a template (dynamic.yml.tmpl) rendered via
  INTEGRATION_REAL_BASEURL (default https://llmgw.example.com), not a
  hardcoded-only hostname.

See task-15-report.md's addendum for full verification detail.
This commit is contained in:
2026-08-20 13:46:45 +01:00
parent 7939c3e74b
commit 730583d98e
8 changed files with 201 additions and 28 deletions
+123 -14
View File
@@ -176,12 +176,15 @@ func TestPluginLoads(t *testing.T) {
}
}
// TestUnifiedChatOpenAIAndAnthropic covers integration test 2: a unified
// chat completion against the mock OpenAI upstream (passed through
// verbatim) and against the mock Anthropic upstream (translated into the
// same OpenAI-shaped envelope, with the client's requested provider-alias
// echoed back in "model").
func TestUnifiedChatOpenAIAndAnthropic(t *testing.T) {
// TestUnifiedChatAllProviders covers integration test 2: a unified chat
// completion against each of the three mock upstreams under real Traefik.
// openai's response is passed through verbatim (already OpenAI-shaped);
// anthropic's and gemini's are translated into the same OpenAI-shaped
// envelope, each with the client's requested provider-alias echoed back
// in "model" — both translation paths run under real Yaegi here, not just
// go test, which is exactly what caught Task 15's yaegi-only bugs in
// registry.go, routes_unified.go, sse.go, and translate_gemini.go.
func TestUnifiedChatAllProviders(t *testing.T) {
openaiReq := map[string]any{
"model": "openai/gpt-mock",
"messages": []map[string]any{{"role": "user", "content": "hi"}},
@@ -208,15 +211,51 @@ func TestUnifiedChatOpenAIAndAnthropic(t *testing.T) {
if model, _ := body2["model"].(string); model != "anthropic/claude-mock" {
t.Errorf("anthropic chat: response model = %q, want the client-requested alias %q echoed back", model, "anthropic/claude-mock")
}
geminiReq := map[string]any{
"model": "gemini/gemini-mock",
"messages": []map[string]any{{"role": "user", "content": "hi"}},
}
resp3, body3 := doJSON(t, http.MethodPost, traefik1URL+"/v1/chat/completions", aliceKey, geminiReq)
if resp3.StatusCode != http.StatusOK {
t.Fatalf("gemini chat: status = %d, body=%#v", resp3.StatusCode, body3)
}
if content := firstChoiceContent(t, body3); content != "mock gemini response" {
t.Errorf("gemini chat: content = %q, want %q", content, "mock gemini response")
}
if model, _ := body3["model"].(string); model != "gemini/gemini-mock" {
t.Errorf("gemini chat: response model = %q, want the client-requested alias %q echoed back", model, "gemini/gemini-mock")
}
}
// TestStreamingIncrementalityProbe covers integration test 3. Per the
// design spec's Appendix A spike lesson, streaming timing must never be
// judged through the Docker Desktop host port-proxy (it buffers) — so this
// runs the timing assertion as a one-shot container inside the compose
// network (`docker compose run --rm probe`), reading its exit code rather
// than making any timing claim from this (host-side) process itself.
// TestStreamingIncrementalityProbe covers integration test 3.
//
// Content correctness always runs, via assertStreamingContent: a host-side
// client reading the full SSE body is valid for CONTENT regardless of how
// many TCP segments the response arrived as — only a TIMING claim is
// invalidated by output buffering, whether that's the Docker Desktop host
// port-proxy (the design spec's Appendix A spike lesson) or, as Task 15
// discovered, Traefik/Yaegi itself. It verifies all 5 of the mock's delta
// chunks arrive in order, followed by the terminal "[DONE]" event.
//
// The true incremental-delivery TIMING assertion — the raw-TCP in-network
// probe, requiring ≥3 distinct inter-arrival gaps of 100ms or more — is
// gated behind INTEGRATION_STREAMING=1 and skipped otherwise. It is
// currently known to fail under real Traefik+Yaegi: a confirmed, external,
// still-open upstream bug (traefik/traefik#10269, labeled
// "kind/bug/confirmed" by Traefik's own maintainers; see also
// traefik/yaegi#1600) means Yaegi cannot detect http.Flusher support on an
// http.ResponseWriter crossing the compiled-to-interpreted boundary, so no
// code in this plugin can force a real per-chunk flush — see sse.go's
// newSSEWriter and task-15-report.md for the full reproduction. Set
// INTEGRATION_STREAMING=1 to re-check once that upstream issue is fixed.
func TestStreamingIncrementalityProbe(t *testing.T) {
assertStreamingContent(t)
if os.Getenv("INTEGRATION_STREAMING") != "1" {
t.Skipf("INTEGRATION_STREAMING not set to 1; skipping the true incremental-delivery timing assertion — known broken under real Traefik+Yaegi (traefik/traefik#10269), not a plugin defect. Content correctness above still ran and passed. Set INTEGRATION_STREAMING=1 to re-check once upstream fixes it.")
}
out, err := exec.Command("docker", "compose", "-f", composeFile(), "run", "--rm", "probe").CombinedOutput()
t.Logf("probe output:\n%s", out)
if err != nil {
@@ -224,11 +263,78 @@ func TestStreamingIncrementalityProbe(t *testing.T) {
}
}
// assertStreamingContent issues one streaming chat completion against the
// mock openai upstream from the host side and verifies its SSE body
// carries the mock's 5 delta chunks — "Hel", "lo", " wor", "ld", "!" — in
// order, followed by the terminal "[DONE]" event.
func assertStreamingContent(t *testing.T) {
t.Helper()
reqBody := map[string]any{
"model": "openai/gpt-mock",
"stream": true,
"messages": []map[string]any{{"role": "user", "content": "hi"}},
}
b, err := json.Marshal(reqBody)
if err != nil {
t.Fatalf("marshal request: %v", err)
}
req, err := http.NewRequest(http.MethodPost, traefik1URL+"/v1/chat/completions", bytes.NewReader(b))
if err != nil {
t.Fatalf("new request: %v", err)
}
req.Header.Set("Authorization", "Bearer "+aliceKey)
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("streaming content request: %v", err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
t.Fatalf("streaming content: status = %d", resp.StatusCode)
}
raw, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read streaming body: %v", err)
}
body := string(raw)
searchFrom := 0
for _, want := range []string{"Hel", "lo", " wor", "ld", "!"} {
marker := `"content":"` + want + `"`
idx := strings.Index(body[searchFrom:], marker)
if idx < 0 {
t.Fatalf("streaming content: delta %q not found in order at or after byte %d of body:\n%s", want, searchFrom, body)
}
searchFrom += idx + len(marker)
}
if !strings.Contains(body[searchFrom:], "[DONE]") {
t.Fatalf("streaming content: [DONE] not found after the final delta in body:\n%s", body)
}
}
// flushRedis clears every key in the shared Redis so a rate-limit test
// starts from a known-empty state. Required for idempotence when re-run
// against a stack `make integration-keep` left up from a previous run —
// without this, a stale counter left over from that earlier run could
// make the very first request of this run already report 429.
func flushRedis(t *testing.T) {
t.Helper()
out, err := exec.Command("docker", "compose", "-f", composeFile(), "exec", "-T", "redis", "redis-cli", "FLUSHALL").CombinedOutput()
if err != nil {
t.Fatalf("flush redis: %v\n%s", err, out)
}
}
// TestCrossReplicaLimits covers integration test 4: bob's limited group
// (requestsPerMinute: 3) is enforced globally across both Traefik
// "replicas" via the Redis they share, regardless of which one a given
// request lands on.
func TestCrossReplicaLimits(t *testing.T) {
flushRedis(t)
reqBody := map[string]any{
"model": "openai/gpt-mock",
"messages": []map[string]any{{"role": "user", "content": "hi"}},
@@ -296,7 +402,10 @@ func TestUsersFileHotReload(t *testing.T) {
t.Fatalf("write users.json: %v", err)
}
deadline := time.Now().Add(10 * time.Second)
// 20s, not the plugin's 5s reload throttle alone: this is a real docker
// compose stack, not an in-process unit test, and needs margin for
// bind-mount propagation latency and scheduler jitter on a loaded host.
deadline := time.Now().Add(20 * time.Second)
var lastStatus int
for time.Now().Before(deadline) {
resp, _ := doJSON(t, http.MethodGet, traefik1URL+"/v1/models", carolKey, nil)
@@ -306,7 +415,7 @@ func TestUsersFileHotReload(t *testing.T) {
}
time.Sleep(time.Second)
}
t.Fatalf("carol's key did not become valid within 10s of users.json changing (last status=%d)", lastStatus)
t.Fatalf("carol's key did not become valid within 20s of users.json changing (last status=%d)", lastStatus)
}
// TestPassthroughAndMCPStream covers integration test 6: native passthrough
+16 -5
View File
@@ -341,6 +341,7 @@ func runProbe(target, apiKey, model string) int {
}
var arrivals []time.Time
var headerBuf []byte
buf := make([]byte, 4096)
headerDone := false
for {
@@ -348,12 +349,22 @@ func runProbe(target, apiKey, model string) int {
if n > 0 {
chunk := buf[:n]
if !headerDone {
if idx := bytes.Index(chunk, []byte("\r\n\r\n")); idx >= 0 {
headerDone = true
chunk = chunk[idx+4:]
} else {
chunk = nil
// The "\r\n\r\n" header/body terminator is not guaranteed to
// land inside a single Read: accumulate every read into a
// rolling buffer and search that, not just the latest
// chunk, so a terminator split across two reads is still
// found instead of silently discarding both reads' bytes.
headerBuf = append(headerBuf, chunk...)
idx := bytes.Index(headerBuf, []byte("\r\n\r\n"))
if idx < 0 {
if readErr != nil {
break
}
continue
}
headerDone = true
chunk = headerBuf[idx+4:]
headerBuf = nil
}
if len(chunk) > 0 && bytes.Contains(chunk, []byte("data:")) {
arrivals = append(arrivals, time.Now())
@@ -73,7 +73,7 @@ http:
providers:
uni:
type: openai
baseUrl: "https://llmgw.example.com"
baseUrl: "__INTEGRATION_REAL_BASEURL__"
apiKey: ""
# Explicit rather than discovered: resolution must not depend
# on this container reaching the real upstream at Traefik