test(e2e): retry post-recovery requests in decode-pod disruption spec (llm-d#1961)

* test(e2e): retry post-recovery requests in decode-pod disruption spec

The final recovery check sent unretried requests seconds after the EPP
deployment turned Available. Envoy's active gRPC health check of the
ext_proc cluster (10s interval) can lag actual EPP readiness in that
window, producing a local 500 with no healthy upstream and failing the
spec despite recovery having succeeded. Wrap the check in Eventually,
requiring three consecutive successes, consistent with the other
disruption specs.

Signed-off-by: nilig <nili.ifergan@gmail.com>

* test(e2e): share the post-recovery probe and verify completion content

The streaming disruption spec had the same unretried post-recovery loop
and the same exposure to transient gateway 500s. Both specs now poll a
shared probe with MustPassRepeatedly(3). tryCompletion verifies finish
reason and echoed text, matching runCompletion, so retry tolerance does
not weaken what the recovery check asserts.

Signed-off-by: nilig <nili.ifergan@gmail.com>

---------

Signed-off-by: nilig <nili.ifergan@gmail.com>

Author: nilig

test/e2e/disruption_test.go

@@ -83,6 +83,19 @@ func eppPodReady(oldPodName string) func() bool {
 	}
 }
 
+// completionRoutedToNamespace sends one completion and reports an error on any
+// failure or namespace-header mismatch, for use inside Eventually blocks.
+func completionRoutedToNamespace() error {
+	nsHdr, _, err := tryCompletion(simplePrompt, simModelName)
+	if err != nil {
+		return err
+	}
+	if nsHdr != nsName {
+		return fmt.Errorf("expected namespace %q, got %q", nsName, nsHdr)
+	}
+	return nil
+}
+
 var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disruptiveTestLabel), func() {
 	ginkgo.When("A decode pod is killed mid-request", func() {
 		ginkgo.It("should recover and route to surviving pods", func() {
@@ -112,7 +125,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 
 			ginkgo.By("Verifying new requests eventually route to a pod other than the killed one")
 			gomega.Eventually(func() error {
-				nsHdr, podHdr, _, err := tryCompletion(simplePrompt, simModelName)
+				nsHdr, podHdr, err := tryCompletion(simplePrompt, simModelName)
 				if err != nil {
 					return err
 				}
@@ -131,11 +144,9 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 				return len(currentDecode)
 			}, readyTimeout, 2*time.Second).Should(gomega.Equal(2))
 
-			ginkgo.By("Verifying requests succeed after recovery")
-			for range 3 {
-				nsHdr, _, _ = runCompletion(simplePrompt, simModelName)
-				gomega.Expect(nsHdr).Should(gomega.Equal(nsName))
-			}
+			ginkgo.By("Verifying requests succeed consistently after recovery")
+			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).
+				MustPassRepeatedly(3).Should(gomega.Succeed())
 		})
 	})
 
@@ -183,11 +194,9 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 				return len(currentDecode)
 			}, readyTimeout, 2*time.Second).Should(gomega.Equal(2))
 
-			ginkgo.By("Verifying requests succeed after recovery")
-			for range 3 {
-				nsHdr, _, _ := runCompletion(simplePrompt, simModelName)
-				gomega.Expect(nsHdr).Should(gomega.Equal(nsName))
-			}
+			ginkgo.By("Verifying requests succeed consistently after recovery")
+			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).
+				MustPassRepeatedly(3).Should(gomega.Succeed())
 		})
 	})
 

test/e2e/requests_test.go

@@ -118,7 +118,7 @@ func runCompletion(prompt string, theModel openai.CompletionNewParamsModel) (str
 
 // tryCompletion is like runCompletion but returns an error instead of asserting,
 // intended for use inside Eventually blocks where transient failures are acceptable.
-func tryCompletion(prompt string, theModel openai.CompletionNewParamsModel) (string, string, string, error) {
+func tryCompletion(prompt string, theModel openai.CompletionNewParamsModel) (string, string, error) {
 	var httpResp *http.Response
 	completionParams := openai.CompletionNewParams{
 		Prompt: openai.CompletionNewParamsPromptUnion{OfString: openai.String(prompt)},
@@ -131,16 +131,23 @@ func tryCompletion(prompt string, theModel openai.CompletionNewParamsModel) (str
 		option.WithRequestTimeout(readyTimeout),
 	)
 	if err != nil {
-		return "", "", "", err
+		return "", "", err
 	}
 	if httpResp == nil {
-		return "", "", "", errors.New("missing http response")
+		return "", "", errors.New("missing http response")
 	}
 	if len(resp.Choices) != 1 {
-		return "", "", "", fmt.Errorf("expected 1 choice, got %d", len(resp.Choices))
+		return "", "", fmt.Errorf("expected 1 choice, got %d", len(resp.Choices))
 	}
-	ns, pod, p := extractInferenceHeaders(httpResp)
-	return ns, pod, p, nil
+	if resp.Choices[0].FinishReason != openai.CompletionChoiceFinishReasonStop {
+		return "", "", fmt.Errorf("expected finish reason %q, got %q",
+			openai.CompletionChoiceFinishReasonStop, resp.Choices[0].FinishReason)
+	}
+	if resp.Choices[0].Text != prompt {
+		return "", "", fmt.Errorf("expected echoed prompt, got %q", resp.Choices[0].Text)
+	}
+	ns, pod, _ := extractInferenceHeaders(httpResp)
+	return ns, pod, nil
 }
 
 func runChatCompletion(prompt, modelName string) (string, string, string) {