test: refactor integration test harness to support data layer testing

cmd/epp/runner/test_runner.go

@@ -18,17 +18,32 @@ package runner
 
 import (
 	"context"
+	"encoding/json"
 
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	fwkdl "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
+	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 )
 
-// NewTestRunnerSetup creates a setup runner dedicated for integration test and its corresponding dataStore.
-func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, pmc backendmetrics.PodMetricsClient) (ctrl.Manager, datastore.Datastore, error) {
+// NewTestRunnerSetup creates a setup runner dedicated for integration tests. When mockDataSource is
+// non-nil, its plugin type is registered as a factory that returns the provided instance, so the
+// YAML config can reference it by type name and the runner wires it into the endpoint factory
+// automatically. Pass nil to fall back to the legacy metrics system with pmc.
+func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, pmc backendmetrics.PodMetricsClient, mockDataSource fwkdl.DataSource) (ctrl.Manager, datastore.Datastore, error) {
 	runner := NewRunner()
 	runner.testOverrideSkipNameValidation = true
+
+	if mockDataSource != nil {
+		mockType := mockDataSource.TypedName().Type
+		fwkplugin.Register(mockType, func(name string, _ json.RawMessage, _ fwkplugin.Handle) (fwkplugin.Plugin, error) {
+			return mockDataSource, nil
+		})
+		defer delete(fwkplugin.Registry, mockType)
+	}
+
 	return runner.setup(ctx, cfg, opts, pmc)
 }

test/integration/epp/util.go → test/integration/epp/common_tests.go

@@ -30,6 +30,14 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/test/integration"
 )
 
+// Model name constants shared across test suites.
+const (
+	modelMyModel       = "my-model"
+	modelMyModelTarget = "my-model-12345"
+	modelSQLLora       = "sql-lora"
+	modelSQLLoraTarget = "sql-lora-1fdg2"
+)
+
 // --- Domain Request Builders ---
 
 // ReqSubset creates a request sequence with Envoy Endpoint Metadata.
@@ -251,6 +259,74 @@ func ExpectGRPCStreamResp(chunks ...string) []*extProcPb.ProcessingResponse {
 	return res
 }
 
+// --- Shared Test Cases ---
+
+// testCase defines a single integration test scenario that can be shared across test suites.
+type testCase struct {
+	name          string
+	requests      []*extProcPb.ProcessingRequest
+	pods          []podState
+	wantResponses []*extProcPb.ProcessingResponse
+	wantMetrics   map[string]string
+	waitForModel  string
+	// requiresCRDs indicates that this test case relies on specific Gateway API CRD features (like
+	// InferenceModelRewrite) which are not available in Standalone runMode without CRD.
+	requiresCRDs bool
+	// wantSpans lists the span names expected to be recorded (hermetic tests only).
+	wantSpans []string
+}
+
+// commonTestCases returns the test cases shared between the standard and data layer test suites.
+// prio adjusts expected priority label values based on execution context (e.g. 0 in NoCRD mode).
+func commonTestCases(prio func(int) int) []testCase {
+	return []testCase{
+		{
+			name:     "select lower queue and kv cache",
+			requests: integration.ReqLLM(logger, "test1", modelMyModel, modelMyModelTarget),
+			pods: []podState{
+				P(0, 3, 0.2),
+				P(1, 0, 0.1), // Winner (Low Queue, Low KV)
+				P(2, 10, 0.2),
+			},
+			wantResponses: ExpectRouteTo("192.168.1.2:8000", modelMyModelTarget, "test1"),
+			wantMetrics: map[string]string{
+				"inference_objective_request_total": cleanMetric(metricReqTotal(modelMyModel, modelMyModelTarget, prio(2))),
+				"inference_pool_ready_pods":         cleanMetric(metricReadyPods(3)),
+			},
+			wantSpans: []string{"gateway.request", "gateway.request_orchestration"},
+		},
+		{
+			name:     "select active lora, low queue",
+			requests: integration.ReqLLM(logger, "test2", modelSQLLora, modelSQLLoraTarget),
+			pods: []podState{
+				P(0, 0, 0.2, "foo", "bar"),
+				P(1, 0, 0.1, "foo", modelSQLLoraTarget), // Winner (Has LoRA)
+				P(2, 10, 0.2, "foo", "bar"),
+			},
+			wantResponses: ExpectRouteTo("192.168.1.2:8000", modelSQLLoraTarget, "test2"),
+			wantMetrics: map[string]string{
+				"inference_objective_request_total": cleanMetric(metricReqTotal(modelSQLLora, modelSQLLoraTarget, prio(2))),
+			},
+		},
+		{
+			name:     "no backend pods available",
+			requests: integration.ReqHeaderOnly(map[string]string{"content-type": "application/json"}),
+			pods:     nil,
+			wantResponses: ExpectReject(envoyTypePb.StatusCode_InternalServerError,
+				"inference error: Internal - no pods available in datastore"),
+		},
+		{
+			name: "request missing model field",
+			requests: integration.ReqRaw(
+				map[string]string{"content-type": "application/json"},
+				`{"prompt":"hello world"}`,
+			),
+			wantResponses: ExpectReject(envoyTypePb.StatusCode_BadRequest,
+				"inference error: BadRequest - model not found in request body"),
+		},
+	}
+}
+
 // --- Data Structures & Metrics Helpers ---
 
 type podState struct {

test/integration/epp/datalayer_integration_test.go

@@ -0,0 +1,66 @@
+/*
+Copyright 2025 The Kubernetes 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.
+*/
+
+package epp
+
+import (
+	"context"
+	"testing"
+
+	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
+
+	"sigs.k8s.io/gateway-api-inference-extension/test/integration"
+)
+
+// TestFullDuplexStreamed_DataLayer runs integration tests through the datalayer metrics pipeline.
+// It mirrors a representative subset of hermetic_test.go test cases but uses the data layer
+// (mock DataSource) instead of the standard FakePodMetricsClient.
+func TestFullDuplexStreamed_DataLayer(t *testing.T) {
+	// Datalayer tests always run in standard mode with base resources (priority=2 in InferenceObjectives).
+	tests := commonTestCases(func(p int) int { return p })
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			h := NewTestHarness(t, ctx, WithStandardMode(), WithDataLayer())
+			h.WithBaseResources().WithPods(tc.pods).WaitForSync(len(tc.pods), modelMyModel)
+			if len(tc.pods) > 0 {
+				h.WaitForReadyPodsMetric(len(tc.pods))
+			}
+
+			responses, err := integration.StreamedRequest(t, h.Client, tc.requests, len(tc.wantResponses))
+			require.NoError(t, err)
+
+			if diff := cmp.Diff(tc.wantResponses, responses,
+				protocmp.Transform(),
+				protocmp.SortRepeated(func(a, b *configPb.HeaderValueOption) bool {
+					return a.GetHeader().GetKey() < b.GetHeader().GetKey()
+				}),
+			); diff != "" {
+				t.Errorf("Response mismatch (-want +got): %v", diff)
+			}
+
+			if len(tc.wantMetrics) > 0 {
+				h.ExpectMetrics(tc.wantMetrics)
+			}
+		})
+	}
+}