Repository: kubernetes-sigs/gateway-api-inference-extension

Issue:
Ensure tests can run with a configurable data layer implementation
Several tests files (e.g., datastore/datastore_test.go, requestcontrol/director_test.go) are using `backendmetrics` directly - and should be modified to run with the configured implementation (either `pkg/epp/backend/metrics` or `pkg/epp/datalayer` - ideally we can run each test with both).

/kind cleanup


Discussion:
@elevran: /milestone Data Layer Foundational Implementation

Solution patch:
diff --git a/pkg/epp/datalayer/fake.go b/pkg/epp/datalayer/fake.go
new file mode 100644
index 0000000..24d01c1
--- /dev/null
+++ b/pkg/epp/datalayer/fake.go
@@ -0,0 +1,58 @@
+/*
+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 datalayer
+
+import (
+	"context"
+	"sync/atomic"
+
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
+)
+
+const (
+	fakeSource = "fake-data-source"
+)
+
+type FakeDataSource struct {
+	typedName *plugins.TypedName
+	callCount int64
+	Metrics   map[types.NamespacedName]*Metrics
+	Errors    map[types.NamespacedName]error
+}
+
+func (fds *FakeDataSource) TypedName() plugins.TypedName {
+	if fds.typedName != nil {
+		return *fds.typedName
+	}
+	return plugins.TypedName{
+		Type: fakeSource,
+		Name: fakeSource,
+	}
+}
+func (fds *FakeDataSource) Extractors() []string           { return []string{} }
+func (fds *FakeDataSource) AddExtractor(_ Extractor) error { return nil }
+
+func (fds *FakeDataSource) Collect(ctx context.Context, ep Endpoint) error {
+	atomic.AddInt64(&fds.callCount, 1)
+	if metrics, ok := fds.Metrics[ep.GetPod().Clone().NamespacedName]; ok {
+		if _, ok := fds.Errors[ep.GetPod().Clone().NamespacedName]; !ok {
+			ep.UpdateMetrics(metrics)
+		}
+	}
+	return nil
+}



Test patch:
diff --git a/pkg/epp/controller/inferencemodelrewrite_reconciler_test.go b/pkg/epp/controller/inferencemodelrewrite_reconciler_test.go
index d03278f..a09f894 100644
--- a/pkg/epp/controller/inferencemodelrewrite_reconciler_test.go
+++ b/pkg/epp/controller/inferencemodelrewrite_reconciler_test.go
@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/apix/v1alpha2"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/common"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	poolutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/pool"
 	utiltest "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
@@ -167,58 +168,64 @@ func TestInferenceModelRewriteReconciler(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			scheme := runtime.NewScheme()
-			_ = clientgoscheme.AddToScheme(scheme)
-			_ = v1alpha2.Install(scheme)
-			_ = v1.Install(scheme)
-			initObjs := []client.Object{}
-			if test.rewrite != nil {
-				initObjs = append(initObjs, test.rewrite)
-			}
-			for _, r := range test.rewritesInAPIServer {
-				initObjs = append(initObjs, r)
-			}
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(initObjs...).
-				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := datastore.NewDatastore(t.Context(), pmf, 0)
-			for _, r := range test.rewritesInStore {
-				ds.ModelRewriteSet(r)
-			}
-			endpointPool := poolutil.InferencePoolToEndpointPool(poolForRewrite)
-			_ = ds.PoolSet(context.Background(), fakeClient, endpointPool)
-			reconciler := &InferenceModelRewriteReconciler{
-				Reader:    fakeClient,
-				Datastore: ds,
-				PoolGKNN: common.GKNN{
-					NamespacedName: types.NamespacedName{Name: poolForRewrite.Name, Namespace: poolForRewrite.Namespace},
-					GroupKind:      schema.GroupKind{Group: poolForRewrite.GroupVersionKind().Group, Kind: poolForRewrite.GroupVersionKind().Kind},
-				},
-			}
-			if test.incomingReq == nil {
-				test.incomingReq = &types.NamespacedName{Name: test.rewrite.Name, Namespace: test.rewrite.Namespace}
-			}
-
-			result, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: *test.incomingReq})
-			if err != nil {
-				t.Fatalf("expected no error, got %v", err)
-			}
-
-			if diff := cmp.Diff(result, test.wantResult); diff != "" {
-				t.Errorf("Unexpected result diff (+got/-want): %s", diff)
-			}
-
-			if len(test.wantRewrites) != len(ds.ModelRewriteGetAll()) {
-				t.Errorf("Unexpected number of rewrites; want: %d, got:%d", len(test.wantRewrites), len(ds.ModelRewriteGetAll()))
-			}
-
-			if diff := diffStoreRewrites(ds, test.wantRewrites); diff != "" {
-				t.Errorf("Unexpected diff (+got/-want): %s", diff)
-			}
-		})
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = v1alpha2.Install(scheme)
+				_ = v1.Install(scheme)
+				initObjs := []client.Object{}
+				if test.rewrite != nil {
+					initObjs = append(initObjs, test.rewrite)
+				}
+				for _, r := range test.rewritesInAPIServer {
+					initObjs = append(initObjs, r)
+				}
+				fakeClient := fake.NewClientBuilder().
+					WithScheme(scheme).
+					WithObjects(initObjs...).
+					Build()
+				ds := datastore.NewDatastore(t.Context(), epf, 0)
+				for _, r := range test.rewritesInStore {
+					ds.ModelRewriteSet(r)
+				}
+				endpointPool := poolutil.InferencePoolToEndpointPool(poolForRewrite)
+				_ = ds.PoolSet(context.Background(), fakeClient, endpointPool)
+				reconciler := &InferenceModelRewriteReconciler{
+					Reader:    fakeClient,
+					Datastore: ds,
+					PoolGKNN: common.GKNN{
+						NamespacedName: types.NamespacedName{Name: poolForRewrite.Name, Namespace: poolForRewrite.Namespace},
+						GroupKind:      schema.GroupKind{Group: poolForRewrite.GroupVersionKind().Group, Kind: poolForRewrite.GroupVersionKind().Kind},
+					},
+				}
+				if test.incomingReq == nil {
+					test.incomingReq = &types.NamespacedName{Name: test.rewrite.Name, Namespace: test.rewrite.Namespace}
+				}
+
+				result, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: *test.incomingReq})
+				if err != nil {
+					t.Fatalf("expected no error, got %v", err)
+				}
+
+				if diff := cmp.Diff(result, test.wantResult); diff != "" {
+					t.Errorf("Unexpected result diff (+got/-want): %s", diff)
+				}
+
+				if len(test.wantRewrites) != len(ds.ModelRewriteGetAll()) {
+					t.Errorf("Unexpected number of rewrites; want: %d, got:%d", len(test.wantRewrites), len(ds.ModelRewriteGetAll()))
+				}
+
+				if diff := diffStoreRewrites(ds, test.wantRewrites); diff != "" {
+					t.Errorf("Unexpected diff (+got/-want): %s", diff)
+				}
+			})
+		}
 	}
 }
 
diff --git a/pkg/epp/controller/inferenceobjective_reconciler_test.go b/pkg/epp/controller/inferenceobjective_reconciler_test.go
index f8d48ec..b4237b0 100644
--- a/pkg/epp/controller/inferenceobjective_reconciler_test.go
+++ b/pkg/epp/controller/inferenceobjective_reconciler_test.go
@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/apix/v1alpha2"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/common"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	poolutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/pool"
 	utiltest "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
@@ -143,59 +144,65 @@ func TestInferenceObjectiveReconciler(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			// Create a fake client with no InferenceObjective objects.
-			scheme := runtime.NewScheme()
-			_ = clientgoscheme.AddToScheme(scheme)
-			_ = v1alpha2.Install(scheme)
-			_ = v1.Install(scheme)
-			initObjs := []client.Object{}
-			if test.objective != nil {
-				initObjs = append(initObjs, test.objective)
-			}
-			for _, m := range test.objectivesInAPIServer {
-				initObjs = append(initObjs, m)
-			}
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(initObjs...).
-				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := datastore.NewDatastore(t.Context(), pmf, 0)
-			for _, m := range test.objectivessInStore {
-				ds.ObjectiveSet(m)
-			}
-			endpointPool := poolutil.InferencePoolToEndpointPool(inferencePool)
-			_ = ds.PoolSet(context.Background(), fakeClient, endpointPool)
-			reconciler := &InferenceObjectiveReconciler{
-				Reader:    fakeClient,
-				Datastore: ds,
-				PoolGKNN: common.GKNN{
-					NamespacedName: types.NamespacedName{Name: inferencePool.Name, Namespace: inferencePool.Namespace},
-					GroupKind:      schema.GroupKind{Group: inferencePool.GroupVersionKind().Group, Kind: inferencePool.GroupVersionKind().Kind},
-				},
-			}
-			if test.incomingReq == nil {
-				test.incomingReq = &types.NamespacedName{Name: test.objective.Name, Namespace: test.objective.Namespace}
-			}
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				// Create a fake client with no InferenceObjective objects.
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = v1alpha2.Install(scheme)
+				_ = v1.Install(scheme)
+				initObjs := []client.Object{}
+				if test.objective != nil {
+					initObjs = append(initObjs, test.objective)
+				}
+				for _, m := range test.objectivesInAPIServer {
+					initObjs = append(initObjs, m)
+				}
+				fakeClient := fake.NewClientBuilder().
+					WithScheme(scheme).
+					WithObjects(initObjs...).
+					Build()
+				ds := datastore.NewDatastore(t.Context(), epf, 0)
+				for _, m := range test.objectivessInStore {
+					ds.ObjectiveSet(m)
+				}
+				endpointPool := poolutil.InferencePoolToEndpointPool(inferencePool)
+				_ = ds.PoolSet(context.Background(), fakeClient, endpointPool)
+				reconciler := &InferenceObjectiveReconciler{
+					Reader:    fakeClient,
+					Datastore: ds,
+					PoolGKNN: common.GKNN{
+						NamespacedName: types.NamespacedName{Name: inferencePool.Name, Namespace: inferencePool.Namespace},
+						GroupKind:      schema.GroupKind{Group: inferencePool.GroupVersionKind().Group, Kind: inferencePool.GroupVersionKind().Kind},
+					},
+				}
+				if test.incomingReq == nil {
+					test.incomingReq = &types.NamespacedName{Name: test.objective.Name, Namespace: test.objective.Namespace}
+				}
 
-			// Call Reconcile.
-			result, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: *test.incomingReq})
-			if err != nil {
-				t.Fatalf("expected no error when resource is not found, got %v", err)
-			}
+				// Call Reconcile.
+				result, err := reconciler.Reconcile(context.Background(), ctrl.Request{NamespacedName: *test.incomingReq})
+				if err != nil {
+					t.Fatalf("expected no error when resource is not found, got %v", err)
+				}
 
-			if diff := cmp.Diff(result, test.wantResult); diff != "" {
-				t.Errorf("Unexpected result diff (+got/-want): %s", diff)
-			}
+				if diff := cmp.Diff(result, test.wantResult); diff != "" {
+					t.Errorf("Unexpected result diff (+got/-want): %s", diff)
+				}
 
-			if len(test.wantObjectives) != len(ds.ObjectiveGetAll()) {
-				t.Errorf("Unexpected; want: %d, got:%d", len(test.wantObjectives), len(ds.ObjectiveGetAll()))
-			}
-			if diff := diffStore(ds, diffStoreParams{wantPool: endpointPool, wantObjectives: test.wantObjectives}); diff != "" {
-				t.Errorf("Unexpected diff (+got/-want): %s", diff)
-			}
+				if len(test.wantObjectives) != len(ds.ObjectiveGetAll()) {
+					t.Errorf("Unexpected; want: %d, got:%d", len(test.wantObjectives), len(ds.ObjectiveGetAll()))
+				}
+				if diff := diffStore(ds, diffStoreParams{wantPool: endpointPool, wantObjectives: test.wantObjectives}); diff != "" {
+					t.Errorf("Unexpected diff (+got/-want): %s", diff)
+				}
 
-		})
+			})
+		}
 	}
 }
diff --git a/pkg/epp/controller/inferencepool_reconciler_test.go b/pkg/epp/controller/inferencepool_reconciler_test.go
index b327ca7..4d1f223 100644
--- a/pkg/epp/controller/inferencepool_reconciler_test.go
+++ b/pkg/epp/controller/inferencepool_reconciler_test.go
@@ -88,91 +88,97 @@ func TestInferencePoolReconciler(t *testing.T) {
 	pool2 := utiltest.MakeInferencePool("pool2").Namespace("pool2-ns").EndpointPickerRef("epp-service").ObjRef()
 	pool2.SetGroupVersionKind(gvk)
 
-	// Set up the scheme.
-	scheme := runtime.NewScheme()
-	_ = clientgoscheme.AddToScheme(scheme)
-	_ = v1alpha2.Install(scheme)
-	_ = v1.Install(scheme)
-	initialObjects := []client.Object{pool1, pool2}
-	for i := range pods {
-		initialObjects = append(initialObjects, pods[i])
-	}
-	fakeClient := fake.NewClientBuilder().
-		WithScheme(scheme).
-		WithObjects(initialObjects...).
-		Build()
-
-	// Create a request for the existing resource.
-	namespacedName := types.NamespacedName{Name: pool1.Name, Namespace: pool1.Namespace}
-	gknn := common.GKNN{
-		NamespacedName: namespacedName,
-		GroupKind: schema.GroupKind{
-			Group: pool1.GroupVersionKind().Group,
-			Kind:  pool1.GroupVersionKind().Kind,
-		},
-	}
-	req := ctrl.Request{NamespacedName: namespacedName}
-	ctx := context.Background()
-
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-	ds := datastore.NewDatastore(ctx, pmf, 0)
-	inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
-
-	// Step 1: Inception, only ready pods matching pool1 are added to the store.
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	endpointPool1 := pool.InferencePoolToEndpointPool(pool1)
-	if diff := diffStore(ds, diffStoreParams{wantPool: endpointPool1, wantPods: []string{"pod1-rank-0", "pod2-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	newPool1 := &v1.InferencePool{}
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	newPool1.Spec.Selector = v1.LabelSelector{
-		MatchLabels: map[v1.LabelKey]v1.LabelValue{"app": "vllm_v2"},
-	}
-	if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool update error: %v", err)
-	}
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	newEndpointPool1 := pool.InferencePoolToEndpointPool(newPool1)
-	if diff := diffStore(ds, diffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	// Step 3: update the inferencePool port
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	newPool1.Spec.TargetPorts = []v1.Port{{Number: 9090}}
-	if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool update error: %v", err)
-	}
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	newEndpointPool1 = pool.InferencePoolToEndpointPool(newPool1)
-	if diff := diffStore(ds, diffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	// Step 4: delete the inferencePool to trigger a datastore clear
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	if err := fakeClient.Delete(ctx, newPool1, &client.DeleteOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool delete error: %v", err)
-	}
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	if diff := diffStore(ds, diffStoreParams{wantPods: []string{}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
+	period := time.Second
+	factories := []datalayer.EndpointFactory{
+		backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+		datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+	}
+	for _, epf := range factories {
+		// Set up the scheme.
+		scheme := runtime.NewScheme()
+		_ = clientgoscheme.AddToScheme(scheme)
+		_ = v1alpha2.Install(scheme)
+		_ = v1.Install(scheme)
+		initialObjects := []client.Object{pool1, pool2}
+		for i := range pods {
+			initialObjects = append(initialObjects, pods[i])
+		}
+		fakeClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithObjects(initialObjects...).
+			Build()
+
+		// Create a request for the existing resource.
+		namespacedName := types.NamespacedName{Name: pool1.Name, Namespace: pool1.Namespace}
+		gknn := common.GKNN{
+			NamespacedName: namespacedName,
+			GroupKind: schema.GroupKind{
+				Group: pool1.GroupVersionKind().Group,
+				Kind:  pool1.GroupVersionKind().Kind,
+			},
+		}
+		req := ctrl.Request{NamespacedName: namespacedName}
+		ctx := context.Background()
+
+		ds := datastore.NewDatastore(ctx, epf, 0)
+		inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
+
+		// Step 1: Inception, only ready pods matching pool1 are added to the store.
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		endpointPool1 := pool.InferencePoolToEndpointPool(pool1)
+		if diff := diffStore(ds, diffStoreParams{wantPool: endpointPool1, wantPods: []string{"pod1-rank-0", "pod2-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		newPool1 := &v1.InferencePool{}
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		newPool1.Spec.Selector = v1.LabelSelector{
+			MatchLabels: map[v1.LabelKey]v1.LabelValue{"app": "vllm_v2"},
+		}
+		if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool update error: %v", err)
+		}
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		newEndpointPool1 := pool.InferencePoolToEndpointPool(newPool1)
+		if diff := diffStore(ds, diffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		// Step 3: update the inferencePool port
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		newPool1.Spec.TargetPorts = []v1.Port{{Number: 9090}}
+		if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool update error: %v", err)
+		}
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		newEndpointPool1 = pool.InferencePoolToEndpointPool(newPool1)
+		if diff := diffStore(ds, diffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		// Step 4: delete the inferencePool to trigger a datastore clear
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		if err := fakeClient.Delete(ctx, newPool1, &client.DeleteOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool delete error: %v", err)
+		}
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		if diff := diffStore(ds, diffStoreParams{wantPods: []string{}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
 	}
 }
 
@@ -235,90 +241,96 @@ func TestXInferencePoolReconciler(t *testing.T) {
 	pool1.SetGroupVersionKind(gvk)
 	pool2.SetGroupVersionKind(gvk)
 
-	// Set up the scheme.
-	scheme := runtime.NewScheme()
-	_ = clientgoscheme.AddToScheme(scheme)
-	_ = v1alpha2.Install(scheme)
-	_ = v1.Install(scheme)
-	initialObjects := []client.Object{pool1, pool2}
-	for i := range pods {
-		initialObjects = append(initialObjects, pods[i])
-	}
-	fakeClient := fake.NewClientBuilder().
-		WithScheme(scheme).
-		WithObjects(initialObjects...).
-		Build()
-
-	// Create a request for the existing resource.
-	namespacedName := types.NamespacedName{Name: pool1.Name, Namespace: pool1.Namespace}
-	gknn := common.GKNN{
-		NamespacedName: namespacedName,
-		GroupKind: schema.GroupKind{
-			Group: pool1.GroupVersionKind().Group,
-			Kind:  pool1.GroupVersionKind().Kind,
-		},
-	}
-	req := ctrl.Request{NamespacedName: namespacedName}
-	ctx := context.Background()
-
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-	ds := datastore.NewDatastore(ctx, pmf, 0)
-	inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
-
-	// Step 1: Inception, only ready pods matching pool1 are added to the store.
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	endpointPool1 := pool.AlphaInferencePoolToEndpointPool(pool1)
-	if diff := xDiffStore(ds, xDiffStoreParams{wantPool: endpointPool1, wantPods: []string{"pod1-rank-0", "pod2-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	newPool1 := &v1alpha2.InferencePool{}
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	newPool1.Spec.Selector = map[v1alpha2.LabelKey]v1alpha2.LabelValue{"app": "vllm_v2"}
-	if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool update error: %v", err)
-	}
-
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	newEndpointPool1 := pool.AlphaInferencePoolToEndpointPool(newPool1)
-	if diff := xDiffStore(ds, xDiffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	// Step 3: update the inferencePool port
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	newPool1.Spec.TargetPortNumber = 9090
-	if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool update error: %v", err)
-	}
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	newEndpointPool1 = pool.AlphaInferencePoolToEndpointPool(newPool1)
-	if diff := xDiffStore(ds, xDiffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
-	}
-
-	// Step 4: delete the inferencePool to trigger a datastore clear
-	if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
-		t.Errorf("Unexpected inferencePool get error: %v", err)
-	}
-	if err := fakeClient.Delete(ctx, newPool1, &client.DeleteOptions{}); err != nil {
-		t.Errorf("Unexpected inferencePool delete error: %v", err)
-	}
-	if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
-		t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-	}
-	if diff := xDiffStore(ds, xDiffStoreParams{wantPods: []string{}}); diff != "" {
-		t.Errorf("Unexpected diff (+got/-want): %s", diff)
+	period := time.Second
+	factories := []datalayer.EndpointFactory{
+		backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+		datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+	}
+	for _, epf := range factories {
+		// Set up the scheme.
+		scheme := runtime.NewScheme()
+		_ = clientgoscheme.AddToScheme(scheme)
+		_ = v1alpha2.Install(scheme)
+		_ = v1.Install(scheme)
+		initialObjects := []client.Object{pool1, pool2}
+		for i := range pods {
+			initialObjects = append(initialObjects, pods[i])
+		}
+		fakeClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithObjects(initialObjects...).
+			Build()
+
+		// Create a request for the existing resource.
+		namespacedName := types.NamespacedName{Name: pool1.Name, Namespace: pool1.Namespace}
+		gknn := common.GKNN{
+			NamespacedName: namespacedName,
+			GroupKind: schema.GroupKind{
+				Group: pool1.GroupVersionKind().Group,
+				Kind:  pool1.GroupVersionKind().Kind,
+			},
+		}
+		req := ctrl.Request{NamespacedName: namespacedName}
+		ctx := context.Background()
+
+		ds := datastore.NewDatastore(ctx, epf, 0)
+		inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
+
+		// Step 1: Inception, only ready pods matching pool1 are added to the store.
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		endpointPool1 := pool.AlphaInferencePoolToEndpointPool(pool1)
+		if diff := xDiffStore(ds, xDiffStoreParams{wantPool: endpointPool1, wantPods: []string{"pod1-rank-0", "pod2-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		newPool1 := &v1alpha2.InferencePool{}
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		newPool1.Spec.Selector = map[v1alpha2.LabelKey]v1alpha2.LabelValue{"app": "vllm_v2"}
+		if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool update error: %v", err)
+		}
+
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		newEndpointPool1 := pool.AlphaInferencePoolToEndpointPool(newPool1)
+		if diff := xDiffStore(ds, xDiffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		// Step 3: update the inferencePool port
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		newPool1.Spec.TargetPortNumber = 9090
+		if err := fakeClient.Update(ctx, newPool1, &client.UpdateOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool update error: %v", err)
+		}
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		newEndpointPool1 = pool.AlphaInferencePoolToEndpointPool(newPool1)
+		if diff := xDiffStore(ds, xDiffStoreParams{wantPool: newEndpointPool1, wantPods: []string{"pod5-rank-0"}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
+
+		// Step 4: delete the inferencePool to trigger a datastore clear
+		if err := fakeClient.Get(ctx, req.NamespacedName, newPool1); err != nil {
+			t.Errorf("Unexpected inferencePool get error: %v", err)
+		}
+		if err := fakeClient.Delete(ctx, newPool1, &client.DeleteOptions{}); err != nil {
+			t.Errorf("Unexpected inferencePool delete error: %v", err)
+		}
+		if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
+			t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+		}
+		if diff := xDiffStore(ds, xDiffStoreParams{wantPods: []string{}}); diff != "" {
+			t.Errorf("Unexpected diff (+got/-want): %s", diff)
+		}
 	}
 }
 
diff --git a/pkg/epp/controller/pod_reconciler_test.go b/pkg/epp/controller/pod_reconciler_test.go
index 09c1201..11b5257 100644
--- a/pkg/epp/controller/pod_reconciler_test.go
+++ b/pkg/epp/controller/pod_reconciler_test.go
@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/pool"
 	utiltest "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
@@ -44,8 +45,6 @@ var (
 	basePod2  = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}, Status: corev1.PodStatus{PodIP: "address-2"}}
 	basePod3  = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}, Status: corev1.PodStatus{PodIP: "address-3"}}
 	basePod11 = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}, Status: corev1.PodStatus{PodIP: "address-11"}}
-	pmc       = &backendmetrics.FakePodMetricsClient{}
-	pmf       = backendmetrics.NewPodMetricsFactory(pmc, time.Second)
 )
 
 func TestPodReconciler(t *testing.T) {
@@ -183,43 +182,50 @@ func TestPodReconciler(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			// Set up the scheme.
-			scheme := runtime.NewScheme()
-			_ = clientgoscheme.AddToScheme(scheme)
-			initialObjects := []client.Object{}
-			if test.incomingPod != nil {
-				initialObjects = append(initialObjects, test.incomingPod)
-			}
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(initialObjects...).
-				Build()
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				// Set up the scheme.
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				initialObjects := []client.Object{}
+				if test.incomingPod != nil {
+					initialObjects = append(initialObjects, test.incomingPod)
+				}
+				fakeClient := fake.NewClientBuilder().
+					WithScheme(scheme).
+					WithObjects(initialObjects...).
+					Build()
 
-			// Configure the initial state of the datastore.
-			store := datastore.NewDatastore(t.Context(), pmf, 0)
-			_ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(test.pool))
-			for _, pod := range test.existingPods {
-				store.PodUpdateOrAddIfNotExist(pod)
-			}
+				// Configure the initial state of the datastore.
+				store := datastore.NewDatastore(t.Context(), epf, 0)
+				_ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(test.pool))
+				for _, pod := range test.existingPods {
+					store.PodUpdateOrAddIfNotExist(pod)
+				}
 
-			podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store}
-			if test.req == nil {
-				namespacedName := types.NamespacedName{Name: test.incomingPod.Name, Namespace: test.incomingPod.Namespace}
-				test.req = &ctrl.Request{NamespacedName: namespacedName}
-			}
-			if _, err := podReconciler.Reconcile(context.Background(), *test.req); err != nil {
-				t.Errorf("Unexpected InferencePool reconcile error: %v", err)
-			}
+				podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store}
+				if test.req == nil {
+					namespacedName := types.NamespacedName{Name: test.incomingPod.Name, Namespace: test.incomingPod.Namespace}
+					test.req = &ctrl.Request{NamespacedName: namespacedName}
+				}
+				if _, err := podReconciler.Reconcile(context.Background(), *test.req); err != nil {
+					t.Errorf("Unexpected InferencePool reconcile error: %v", err)
+				}
 
-			var gotPods []*corev1.Pod
-			for _, pm := range store.PodList(datastore.AllPodsPredicate) {
-				pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().PodName, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().GetIPAddress()}}
-				gotPods = append(gotPods, pod)
-			}
-			if !cmp.Equal(gotPods, test.wantPods, cmpopts.SortSlices(func(a, b *corev1.Pod) bool { return a.Name < b.Name })) {
-				t.Errorf("got (%v) != want (%v);", gotPods, test.wantPods)
-			}
-		})
+				var gotPods []*corev1.Pod
+				for _, pm := range store.PodList(datastore.AllPodsPredicate) {
+					pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().PodName, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().GetIPAddress()}}
+					gotPods = append(gotPods, pod)
+				}
+				if !cmp.Equal(gotPods, test.wantPods, cmpopts.SortSlices(func(a, b *corev1.Pod) bool { return a.Name < b.Name })) {
+					t.Errorf("got (%v) != want (%v);", gotPods, test.wantPods)
+				}
+			})
+		}
 	}
 }
diff --git a/pkg/epp/datalayer/collector_test.go b/pkg/epp/datalayer/collector_test.go
index 22408fc..f05175f 100644
--- a/pkg/epp/datalayer/collector_test.go
+++ b/pkg/epp/datalayer/collector_test.go
@@ -27,32 +27,10 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer/mocks"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
 )
 
 // --- Test Stubs ---
 
-type DummySource struct {
-	callCount int64
-}
-
-const (
-	dummySource = "test-dummy-data-source"
-)
-
-func (d *DummySource) TypedName() plugins.TypedName {
-	return plugins.TypedName{
-		Type: dummySource,
-		Name: dummySource,
-	}
-}
-func (d *DummySource) Extractors() []string           { return []string{} }
-func (d *DummySource) AddExtractor(_ Extractor) error { return nil }
-func (d *DummySource) Collect(ctx context.Context, ep Endpoint) error {
-	atomic.AddInt64(&d.callCount, 1)
-	return nil
-}
-
 func defaultEndpoint() Endpoint {
 	pod := &PodInfo{
 		NamespacedName: types.NamespacedName{
@@ -69,7 +47,7 @@ func defaultEndpoint() Endpoint {
 
 var (
 	endpoint = defaultEndpoint()
-	sources  = []DataSource{&DummySource{}}
+	sources  = []DataSource{&FakeDataSource{}}
 )
 
 func TestCollectorCanStartOnlyOnce(t *testing.T) {
@@ -101,7 +79,7 @@ func TestCollectorCanStopOnlyOnce(t *testing.T) {
 }
 
 func TestCollectorCollectsOnTicks(t *testing.T) {
-	source := &DummySource{}
+	source := &FakeDataSource{}
 	c := NewCollector()
 	ticker := mocks.NewTicker()
 	ctx := context.Background()
@@ -119,7 +97,7 @@ func TestCollectorCollectsOnTicks(t *testing.T) {
 }
 
 func TestCollectorStopCancelsContext(t *testing.T) {
-	source := &DummySource{}
+	source := &FakeDataSource{}
 	c := NewCollector()
 	ticker := mocks.NewTicker()
 	ctx := context.Background()
diff --git a/pkg/epp/datalayer/datasource_test.go b/pkg/epp/datalayer/datasource_test.go
index c1f0edc..9b46563 100644
--- a/pkg/epp/datalayer/datasource_test.go
+++ b/pkg/epp/datalayer/datasource_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package datalayer
 
 import (
-	"context"
 	"reflect"
 	"testing"
 
@@ -30,18 +29,9 @@ const (
 	testType = "test-ds-type"
 )
 
-type mockDataSource struct {
-	typedName plugins.TypedName
-}
-
-func (m *mockDataSource) TypedName() plugins.TypedName                { return m.typedName }
-func (m *mockDataSource) Extractors() []string                        { return []string{} }
-func (m *mockDataSource) AddExtractor(_ Extractor) error              { return nil }
-func (m *mockDataSource) Collect(_ context.Context, _ Endpoint) error { return nil }
-
 func TestRegisterAndGetSource(t *testing.T) {
 	reg := DataSourceRegistry{}
-	ds := &mockDataSource{typedName: plugins.TypedName{Type: testType, Name: testType}}
+	ds := &FakeDataSource{typedName: &plugins.TypedName{Type: testType, Name: testType}}
 
 	err := reg.Register(ds)
 	assert.NoError(t, err, "expected no error on first registration")
diff --git a/pkg/epp/datalayer/factory_test.go b/pkg/epp/datalayer/factory_test.go
index 35cce88..302baa3 100644
--- a/pkg/epp/datalayer/factory_test.go
+++ b/pkg/epp/datalayer/factory_test.go
@@ -28,7 +28,7 @@ import (
 )
 
 func TestFactory(t *testing.T) {
-	source := &DummySource{}
+	source := &FakeDataSource{}
 	factory := NewEndpointFactory([]DataSource{source}, 100*time.Millisecond)
 
 	pod1 := &PodInfo{
diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go
index be5b461..71ea2aa 100644
--- a/pkg/epp/datastore/datastore_test.go
+++ b/pkg/epp/datastore/datastore_test.go
@@ -78,35 +78,43 @@ func TestPool(t *testing.T) {
 			wantSynced: false,
 		},
 	}
+
 	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			// Set up the scheme.
-			scheme := runtime.NewScheme()
-			_ = clientgoscheme.AddToScheme(scheme)
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := NewDatastore(context.Background(), pmf, 0)
-			_ = ds.PoolSet(context.Background(), fakeClient, pooltuil.InferencePoolToEndpointPool(tt.inferencePool))
-			gotPool, gotErr := ds.PoolGet()
-			if diff := cmp.Diff(tt.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" {
-				t.Errorf("Unexpected error diff (+got/-want): %s", diff)
-			}
-			if diff := cmp.Diff(pooltuil.InferencePoolToEndpointPool(tt.wantPool), gotPool); diff != "" {
-				t.Errorf("Unexpected pool diff (+got/-want): %s", diff)
-			}
-			gotSynced := ds.PoolHasSynced()
-			if diff := cmp.Diff(tt.wantSynced, gotSynced); diff != "" {
-				t.Errorf("Unexpected synced diff (+got/-want): %s", diff)
-			}
-			if tt.labels != nil {
-				gotLabelsMatch := ds.PoolLabelsMatch(tt.labels)
-				if diff := cmp.Diff(tt.wantLabelsMatch, gotLabelsMatch); diff != "" {
-					t.Errorf("Unexpected labels match diff (+got/-want): %s", diff)
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(tt.name, func(t *testing.T) {
+				// Set up the scheme.
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				fakeClient := fake.NewClientBuilder().
+					WithScheme(scheme).
+					Build()
+
+				ds := NewDatastore(context.Background(), epf, 0)
+				_ = ds.PoolSet(context.Background(), fakeClient, pooltuil.InferencePoolToEndpointPool(tt.inferencePool))
+				gotPool, gotErr := ds.PoolGet()
+				if diff := cmp.Diff(tt.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" {
+					t.Errorf("Unexpected error diff (+got/-want): %s", diff)
+				}
+				if diff := cmp.Diff(pooltuil.InferencePoolToEndpointPool(tt.wantPool), gotPool); diff != "" {
+					t.Errorf("Unexpected pool diff (+got/-want): %s", diff)
+				}
+				gotSynced := ds.PoolHasSynced()
+				if diff := cmp.Diff(tt.wantSynced, gotSynced); diff != "" {
+					t.Errorf("Unexpected synced diff (+got/-want): %s", diff)
+				}
+				if tt.labels != nil {
+					gotLabelsMatch := ds.PoolLabelsMatch(tt.labels)
+					if diff := cmp.Diff(tt.wantLabelsMatch, gotLabelsMatch); diff != "" {
+						t.Errorf("Unexpected labels match diff (+got/-want): %s", diff)
+					}
 				}
-			}
-		})
+			})
+		}
 	}
 }
 
@@ -192,24 +200,30 @@ func TestObjective(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := NewDatastore(t.Context(), pmf, 0)
-			for _, m := range test.existingModels {
-				ds.ObjectiveSet(m)
-			}
-
-			gotOpResult := test.op(ds)
-			if gotOpResult != test.wantOpResult {
-				t.Errorf("Unexpected operation result, want: %v, got: %v", test.wantOpResult, gotOpResult)
-			}
-
-			if diff := cmp.Diff(test.wantModels, ds.ObjectiveGetAll(), cmpopts.SortSlices(func(a, b *v1alpha2.InferenceObjective) bool {
-				return a.Name < b.Name
-			})); diff != "" {
-				t.Errorf("Unexpected models diff: %s", diff)
-			}
-		})
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				ds := NewDatastore(t.Context(), epf, 0)
+				for _, m := range test.existingModels {
+					ds.ObjectiveSet(m)
+				}
+
+				gotOpResult := test.op(ds)
+				if gotOpResult != test.wantOpResult {
+					t.Errorf("Unexpected operation result, want: %v, got: %v", test.wantOpResult, gotOpResult)
+				}
+
+				if diff := cmp.Diff(test.wantModels, ds.ObjectiveGetAll(), cmpopts.SortSlices(func(a, b *v1alpha2.InferenceObjective) bool {
+					return a.Name < b.Name
+				})); diff != "" {
+					t.Errorf("Unexpected models diff: %s", diff)
+				}
+			})
+		}
 	}
 }
 
@@ -266,42 +280,38 @@ var (
 func TestMetrics(t *testing.T) {
 	tests := []struct {
 		name      string
-		pmc       backendmetrics.PodMetricsClient
+		metrics   map[types.NamespacedName]*backendmetrics.MetricsState
+		err       map[types.NamespacedName]error
 		storePods []*corev1.Pod
 		want      []*backendmetrics.MetricsState
 		predict   func(backendmetrics.PodMetrics) bool
 	}{
 		{
 			name: "Probing metrics success",
-			pmc: &backendmetrics.FakePodMetricsClient{
-				Res: map[types.NamespacedName]*backendmetrics.MetricsState{
-					pod1NamespacedName: pod1Metrics,
-					pod2NamespacedName: pod2Metrics,
-				},
+			metrics: map[types.NamespacedName]*backendmetrics.MetricsState{
+				pod1NamespacedName: pod1Metrics,
+				pod2NamespacedName: pod2Metrics,
 			},
 			storePods: []*corev1.Pod{pod1, pod2},
 			want:      []*backendmetrics.MetricsState{pod1Metrics, pod2Metrics},
 		},
 		{
 			name: "Only pods in are probed",
-			pmc: &backendmetrics.FakePodMetricsClient{
-				Res: map[types.NamespacedName]*backendmetrics.MetricsState{
-					pod1NamespacedName: pod1Metrics,
-					pod2NamespacedName: pod2Metrics,
-				},
+			metrics: map[types.NamespacedName]*backendmetrics.MetricsState{
+				pod1NamespacedName: pod1Metrics,
+				pod2NamespacedName: pod2Metrics,
 			},
 			storePods: []*corev1.Pod{pod1},
 			want:      []*backendmetrics.MetricsState{pod1Metrics},
 		},
 		{
 			name: "Probing metrics error",
-			pmc: &backendmetrics.FakePodMetricsClient{
-				Err: map[types.NamespacedName]error{
-					pod2NamespacedName: errors.New("injected error"),
-				},
-				Res: map[types.NamespacedName]*backendmetrics.MetricsState{
-					pod1NamespacedName: pod1Metrics,
-				},
+			err: map[types.NamespacedName]error{
+				pod2NamespacedName: errors.New("injected error"),
+			},
+			metrics: map[types.NamespacedName]*backendmetrics.MetricsState{
+				pod1NamespacedName: pod1Metrics,
+				pod2NamespacedName: pod2Metrics,
 			},
 			storePods: []*corev1.Pod{pod1, pod2},
 			want: []*backendmetrics.MetricsState{pod1Metrics,
@@ -318,37 +328,43 @@ func TestMetrics(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
-			// Set up the scheme.
-			scheme := runtime.NewScheme()
-			_ = clientgoscheme.AddToScheme(scheme)
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(test.pmc, time.Millisecond)
-			ds := NewDatastore(ctx, pmf, 0)
-			_ = ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(inferencePool))
-			for _, pod := range test.storePods {
-				ds.PodUpdateOrAddIfNotExist(pod)
-			}
-			time.Sleep(1 * time.Second) // Give some time for the metrics to be fetched.
-			if test.predict == nil {
-				test.predict = AllPodsPredicate
-			}
-			assert.EventuallyWithT(t, func(t *assert.CollectT) {
-				got := ds.PodList(test.predict)
-				metrics := []*backendmetrics.MetricsState{}
-				for _, one := range got {
-					metrics = append(metrics, one.GetMetrics())
+		period := time.Millisecond
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{Res: test.metrics, Err: test.err}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{Metrics: test.metrics, Errors: test.err}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+				// Set up the scheme.
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				fakeClient := fake.NewClientBuilder().
+					WithScheme(scheme).
+					Build()
+				ds := NewDatastore(ctx, epf, 0)
+				_ = ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(inferencePool))
+				for _, pod := range test.storePods {
+					ds.PodUpdateOrAddIfNotExist(pod)
+				}
+				time.Sleep(1 * time.Second) // Give some time for the metrics to be fetched.
+				if test.predict == nil {
+					test.predict = AllPodsPredicate
 				}
-				diff := cmp.Diff(test.want, metrics, cmpopts.IgnoreFields(backendmetrics.MetricsState{}, "UpdateTime"), cmpopts.SortSlices(func(a, b *backendmetrics.MetricsState) bool {
-					return a.String() < b.String()
-				}))
-				assert.Equal(t, "", diff, "Unexpected diff (+got/-want)")
-			}, 5*time.Second, time.Millisecond)
-		})
+				assert.EventuallyWithT(t, func(t *assert.CollectT) {
+					got := ds.PodList(test.predict)
+					metrics := []*backendmetrics.MetricsState{}
+					for _, one := range got {
+						metrics = append(metrics, one.GetMetrics())
+					}
+					diff := cmp.Diff(test.want, metrics, cmpopts.IgnoreFields(backendmetrics.MetricsState{}, "UpdateTime"), cmpopts.SortSlices(func(a, b *backendmetrics.MetricsState) bool {
+						return a.String() < b.String()
+					}))
+					assert.Equal(t, "", diff, "Unexpected diff (+got/-want)")
+				}, 5*time.Second, time.Millisecond)
+			})
+		}
 	}
 }
 
@@ -393,28 +409,34 @@ func TestPods(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			ctx := context.Background()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := NewDatastore(t.Context(), pmf, 0)
-			fakeClient := fake.NewFakeClient()
-			if err := ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(inferencePool)); err != nil {
-				t.Error(err)
-			}
-			for _, pod := range test.existingPods {
-				ds.PodUpdateOrAddIfNotExist(pod)
-			}
-
-			test.op(ctx, ds)
-			var gotPods []*corev1.Pod
-			for _, pm := range ds.PodList(AllPodsPredicate) {
-				pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().PodName, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().GetIPAddress()}}
-				gotPods = append(gotPods, pod)
-			}
-			if !cmp.Equal(gotPods, test.wantPods, cmpopts.SortSlices(func(a, b *corev1.Pod) bool { return a.Name < b.Name })) {
-				t.Errorf("got (%v) != want (%v);", gotPods, test.wantPods)
-			}
-		})
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				ctx := context.Background()
+				ds := NewDatastore(t.Context(), epf, 0)
+				fakeClient := fake.NewFakeClient()
+				if err := ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(inferencePool)); err != nil {
+					t.Error(err)
+				}
+				for _, pod := range test.existingPods {
+					ds.PodUpdateOrAddIfNotExist(pod)
+				}
+
+				test.op(ctx, ds)
+				var gotPods []*corev1.Pod
+				for _, pm := range ds.PodList(AllPodsPredicate) {
+					pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().PodName, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().GetIPAddress()}}
+					gotPods = append(gotPods, pod)
+				}
+				if !cmp.Equal(gotPods, test.wantPods, cmpopts.SortSlices(func(a, b *corev1.Pod) bool { return a.Name < b.Name })) {
+					t.Errorf("got (%v) != want (%v);", gotPods, test.wantPods)
+				}
+			})
+		}
 	}
 }
 
@@ -577,26 +599,32 @@ func TestPodInfo(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			ctx := context.Background()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-			ds := NewDatastore(t.Context(), pmf, 0)
-			fakeClient := fake.NewFakeClient()
-			if err := ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(test.pool)); err != nil {
-				t.Error(err)
-			}
-			for _, pod := range test.existingPods {
-				ds.PodUpdateOrAddIfNotExist(pod)
-			}
-
-			test.op(ctx, ds)
-			var gotPodInfos []*datalayer.PodInfo
-			for _, pm := range ds.PodList(AllPodsPredicate) {
-				gotPodInfos = append(gotPodInfos, pm.GetPod())
-			}
-			if diff := cmp.Diff(test.wantPodInfos, gotPodInfos, cmpopts.SortSlices(func(a, b *datalayer.PodInfo) bool { return a.NamespacedName.Name < b.NamespacedName.Name })); diff != "" {
-				t.Errorf("ConvertTo() mismatch (-want +got):\n%s", diff)
-			}
-		})
+		period := time.Second
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				ctx := context.Background()
+				ds := NewDatastore(t.Context(), epf, 0)
+				fakeClient := fake.NewFakeClient()
+				if err := ds.PoolSet(ctx, fakeClient, pooltuil.InferencePoolToEndpointPool(test.pool)); err != nil {
+					t.Error(err)
+				}
+				for _, pod := range test.existingPods {
+					ds.PodUpdateOrAddIfNotExist(pod)
+				}
+
+				test.op(ctx, ds)
+				var gotPodInfos []*datalayer.PodInfo
+				for _, pm := range ds.PodList(AllPodsPredicate) {
+					gotPodInfos = append(gotPodInfos, pm.GetPod())
+				}
+				if diff := cmp.Diff(test.wantPodInfos, gotPodInfos, cmpopts.SortSlices(func(a, b *datalayer.PodInfo) bool { return a.NamespacedName.Name < b.NamespacedName.Name })); diff != "" {
+					t.Errorf("ConvertTo() mismatch (-want +got):\n%s", diff)
+				}
+			})
+		}
 	}
 }
diff --git a/pkg/epp/metrics/collectors/inference_pool_test.go b/pkg/epp/metrics/collectors/inference_pool_test.go
index 20ab69f..0b0ff3f 100644
--- a/pkg/epp/metrics/collectors/inference_pool_test.go
+++ b/pkg/epp/metrics/collectors/inference_pool_test.go
@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	poolutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/pool"
 )
@@ -50,54 +51,64 @@ var (
 )
 
 func TestNoMetricsCollected(t *testing.T) {
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-	ds := datastore.NewDatastore(context.Background(), pmf, 0)
-
-	collector := &inferencePoolMetricsCollector{
-		ds: ds,
+	period := time.Second
+	factories := []datalayer.EndpointFactory{
+		backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+		datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
 	}
+	for _, epf := range factories {
+		ds := datastore.NewDatastore(context.Background(), epf, 0)
+
+		collector := &inferencePoolMetricsCollector{
+			ds: ds,
+		}
 
-	if err := testutil.CollectAndCompare(collector, strings.NewReader(""), ""); err != nil {
-		t.Fatal(err)
+		if err := testutil.CollectAndCompare(collector, strings.NewReader(""), ""); err != nil {
+			t.Fatal(err)
+		}
 	}
 }
 
 func TestMetricsCollected(t *testing.T) {
-	pmc := &backendmetrics.FakePodMetricsClient{
-		Res: map[types.NamespacedName]*backendmetrics.MetricsState{
-			pod1NamespacedName: pod1Metrics,
-		},
+	metrics := map[types.NamespacedName]*backendmetrics.MetricsState{
+		pod1NamespacedName: pod1Metrics,
 	}
-	pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond)
-	inferencePool := &v1.InferencePool{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "test-pool",
-		},
-		Spec: v1.InferencePoolSpec{
-			TargetPorts: []v1.Port{{Number: v1.PortNumber(int32(8000))}},
-		},
-	}
-	ds := datastore.NewDatastore(context.Background(), pmf, 0)
-
-	scheme := runtime.NewScheme()
-	fakeClient := fake.NewClientBuilder().
-		WithScheme(scheme).
-		Build()
-
-	_ = ds.PoolSet(context.Background(), fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool))
-	_ = ds.PodUpdateOrAddIfNotExist(pod1)
-
-	time.Sleep(1 * time.Second)
-
-	collector := &inferencePoolMetricsCollector{
-		ds: ds,
+	period := time.Millisecond
+	factories := []datalayer.EndpointFactory{
+		backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{Res: metrics}, period),
+		datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{Metrics: metrics}}, period),
 	}
-	err := testutil.CollectAndCompare(collector, strings.NewReader(`
+	for _, epf := range factories {
+		inferencePool := &v1.InferencePool{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-pool",
+			},
+			Spec: v1.InferencePoolSpec{
+				TargetPorts: []v1.Port{{Number: v1.PortNumber(int32(8000))}},
+			},
+		}
+		ds := datastore.NewDatastore(context.Background(), epf, 0)
+
+		scheme := runtime.NewScheme()
+		fakeClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			Build()
+
+		_ = ds.PoolSet(context.Background(), fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool))
+		_ = ds.PodUpdateOrAddIfNotExist(pod1)
+
+		time.Sleep(1 * time.Second)
+
+		collector := &inferencePoolMetricsCollector{
+			ds: ds,
+		}
+		err := testutil.CollectAndCompare(collector, strings.NewReader(`
 		# HELP inference_pool_per_pod_queue_size [ALPHA] The total number of requests pending in the model server queue for each underlying pod.
 		# TYPE inference_pool_per_pod_queue_size gauge
 		inference_pool_per_pod_queue_size{model_server_pod="pod1-rank-0",name="test-pool"} 100
 `), "inference_pool_per_pod_queue_size")
-	if err != nil {
-		t.Fatal(err)
+		if err != nil {
+			t.Fatal(err)
+		}
 	}
 }
diff --git a/pkg/epp/requestcontrol/director_test.go b/pkg/epp/requestcontrol/director_test.go
index c093654..58372e7 100644
--- a/pkg/epp/requestcontrol/director_test.go
+++ b/pkg/epp/requestcontrol/director_test.go
@@ -260,39 +260,6 @@ func TestDirector_HandleRequest(t *testing.T) {
 		},
 	}
 
-	// Datastore setup
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
-	ds := datastore.NewDatastore(t.Context(), pmf, 0)
-	ds.ObjectiveSet(ioFoodReview)
-	ds.ObjectiveSet(ioFoodReviewResolve)
-	ds.ObjectiveSet(ioFoodReviewSheddable)
-	ds.ModelRewriteSet(rewrite)
-
-	scheme := runtime.NewScheme()
-	_ = clientgoscheme.AddToScheme(scheme)
-	fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
-
-	if err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(pool)); err != nil {
-		t.Fatalf("Error while setting inference pool: %v", err)
-	}
-
-	for i := range 5 {
-		// Pod setup
-		testPod := &corev1.Pod{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      fmt.Sprintf("pod%v", i+1),
-				Namespace: "default",
-				Labels:    map[string]string{"app": "inference"},
-			},
-			Status: corev1.PodStatus{
-				PodIP:      fmt.Sprintf("192.168.%v.100", i+1),
-				Phase:      corev1.PodRunning,
-				Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
-			},
-		}
-		ds.PodUpdateOrAddIfNotExist(testPod)
-	}
-
 	defaultSuccessfulScheduleResults := &schedulingtypes.SchedulingResult{
 		ProfileResults: map[string]*schedulingtypes.ProfileRunResult{
 			"testProfile": {
@@ -646,70 +613,108 @@ func TestDirector_HandleRequest(t *testing.T) {
 		},
 	}
 
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			mockSched := &mockScheduler{}
-			if test.schedulerMockSetup != nil {
-				test.schedulerMockSetup(mockSched)
-			}
-			config := NewConfig()
-			if test.prepareDataPlugin != nil {
-				config = config.WithPrepareDataPlugins(test.prepareDataPlugin)
-			}
-			config = config.WithAdmissionPlugins(newMockAdmissionPlugin("test-admit-plugin", test.admitRequestDenialError))
-
-			locator := NewCachedPodLocator(context.Background(), NewDatastorePodLocator(ds), time.Minute)
-			director := NewDirectorWithConfig(ds, mockSched, test.mockAdmissionController, locator, config)
-			if test.name == "successful request with model rewrite" {
-				mockDs := &mockDatastore{
-					pods:     ds.PodList(datastore.AllPodsPredicate),
-					rewrites: []*v1alpha2.InferenceModelRewrite{rewrite},
-				}
-				director.datastore = mockDs
-				director.podLocator = NewCachedPodLocator(context.Background(), NewDatastorePodLocator(mockDs), time.Minute)
-			}
+	period := time.Second
+	factories := []datalayer.EndpointFactory{
+		backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+		datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+	}
+	for _, epf := range factories {
+		// Datastore setup
+		ds := datastore.NewDatastore(t.Context(), epf, 0)
+		ds.ObjectiveSet(ioFoodReview)
+		ds.ObjectiveSet(ioFoodReviewResolve)
+		ds.ObjectiveSet(ioFoodReviewSheddable)
+		ds.ModelRewriteSet(rewrite)
+
+		scheme := runtime.NewScheme()
+		_ = clientgoscheme.AddToScheme(scheme)
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
+
+		if err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(pool)); err != nil {
+			t.Fatalf("Error while setting inference pool: %v", err)
+		}
 
-			reqCtx := &handlers.RequestContext{
-				Request: &handlers.Request{
-					// Create a copy of the map for each test run to avoid mutation issues.
-					Body: make(map[string]any),
-					Headers: map[string]string{
-						requtil.RequestIdHeaderKey: "test-req-id-" + test.name, // Ensure a default request ID
-					},
+		for i := range 5 {
+			// Pod setup
+			testPod := &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("pod%v", i+1),
+					Namespace: "default",
+					Labels:    map[string]string{"app": "inference"},
+				},
+				Status: corev1.PodStatus{
+					PodIP:      fmt.Sprintf("192.168.%v.100", i+1),
+					Phase:      corev1.PodRunning,
+					Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 				},
-				ObjectiveKey:    test.inferenceObjectiveName,
-				TargetModelName: test.initialTargetModelName,
 			}
-			// Deep copy the body map.
-			maps.Copy(reqCtx.Request.Body, test.reqBodyMap)
+			ds.PodUpdateOrAddIfNotExist(testPod)
+		}
 
-			returnedReqCtx, err := director.HandleRequest(ctx, reqCtx)
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				mockSched := &mockScheduler{}
+				if test.schedulerMockSetup != nil {
+					test.schedulerMockSetup(mockSched)
+				}
+				config := NewConfig()
+				if test.prepareDataPlugin != nil {
+					config = config.WithPrepareDataPlugins(test.prepareDataPlugin)
+				}
+				config = config.WithAdmissionPlugins(newMockAdmissionPlugin("test-admit-plugin", test.admitRequestDenialError))
+
+				locator := NewCachedPodLocator(context.Background(), NewDatastorePodLocator(ds), time.Minute)
+				director := NewDirectorWithConfig(ds, mockSched, test.mockAdmissionController, locator, config)
+				if test.name == "successful request with model rewrite" {
+					mockDs := &mockDatastore{
+						pods:     ds.PodList(datastore.AllPodsPredicate),
+						rewrites: []*v1alpha2.InferenceModelRewrite{rewrite},
+					}
+					director.datastore = mockDs
+					director.podLocator = NewCachedPodLocator(context.Background(), NewDatastorePodLocator(mockDs), time.Minute)
+				}
+				reqCtx := &handlers.RequestContext{
+					Request: &handlers.Request{
+						// Create a copy of the map for each test run to avoid mutation issues.
+						Body: make(map[string]any),
+						Headers: map[string]string{
+							requtil.RequestIdHeaderKey: "test-req-id-" + test.name, // Ensure a default request ID
+						},
+					},
+					ObjectiveKey:    test.inferenceObjectiveName,
+					TargetModelName: test.initialTargetModelName,
+				}
+				// Deep copy the body map.
+				maps.Copy(reqCtx.Request.Body, test.reqBodyMap)
 
-			if test.wantErrCode != "" {
-				assert.Error(t, err, "HandleRequest() should have returned an error")
-				var e errutil.Error
-				if assert.ErrorAs(t, err, &e, "Error should be of type errutil.Error") {
-					assert.Equal(t, test.wantErrCode, e.Code, "Error code mismatch")
+				returnedReqCtx, err := director.HandleRequest(ctx, reqCtx)
+
+				if test.wantErrCode != "" {
+					assert.Error(t, err, "HandleRequest() should have returned an error")
+					var e errutil.Error
+					if assert.ErrorAs(t, err, &e, "Error should be of type errutil.Error") {
+						assert.Equal(t, test.wantErrCode, e.Code, "Error code mismatch")
+					}
+					return
 				}
-				return
-			}
 
-			assert.NoError(t, err, "HandleRequest() returned unexpected error")
+				assert.NoError(t, err, "HandleRequest() returned unexpected error")
 
-			if test.wantReqCtx != nil {
-				assert.Equal(t, test.wantReqCtx.ObjectiveKey, returnedReqCtx.ObjectiveKey, "reqCtx.Model mismatch")
-				assert.Equal(t, test.wantReqCtx.TargetModelName, returnedReqCtx.TargetModelName,
-					"reqCtx.ResolvedTargetModel mismatch")
-				assert.Equal(t, test.wantReqCtx.TargetPod, returnedReqCtx.TargetPod, "reqCtx.TargetPod mismatch")
-				assert.Equal(t, test.wantReqCtx.TargetEndpoint, returnedReqCtx.TargetEndpoint, "reqCtx.TargetEndpoint mismatch")
-			}
+				if test.wantReqCtx != nil {
+					assert.Equal(t, test.wantReqCtx.ObjectiveKey, returnedReqCtx.ObjectiveKey, "reqCtx.Model mismatch")
+					assert.Equal(t, test.wantReqCtx.TargetModelName, returnedReqCtx.TargetModelName,
+						"reqCtx.ResolvedTargetModel mismatch")
+					assert.Equal(t, test.wantReqCtx.TargetPod, returnedReqCtx.TargetPod, "reqCtx.TargetPod mismatch")
+					assert.Equal(t, test.wantReqCtx.TargetEndpoint, returnedReqCtx.TargetEndpoint, "reqCtx.TargetEndpoint mismatch")
+				}
 
-			if test.wantMutatedBodyModel != "" {
-				assert.NotNil(t, returnedReqCtx.Request.Body, "Expected mutated body, but reqCtx.Request.Body is nil")
-				assert.Equal(t, test.wantMutatedBodyModel, returnedReqCtx.Request.Body["model"],
-					"Mutated reqCtx.Request.Body model mismatch")
-			}
-		})
+				if test.wantMutatedBodyModel != "" {
+					assert.NotNil(t, returnedReqCtx.Request.Body, "Expected mutated body, but reqCtx.Request.Body is nil")
+					assert.Equal(t, test.wantMutatedBodyModel, returnedReqCtx.Request.Body["model"],
+						"Mutated reqCtx.Request.Body model mismatch")
+				}
+			})
+		}
 	}
 }
 
@@ -758,27 +763,33 @@ func TestGetRandomPod(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Millisecond)
-			endpointPool := poolutil.InferencePoolToEndpointPool(pool)
-			ds := datastore.NewDatastore(t.Context(), pmf, 0)
-			err := ds.PoolSet(t.Context(), fakeClient, endpointPool)
-			if err != nil {
-				t.Errorf("unexpected error setting pool: %s", err)
-			}
-			for _, pod := range test.storePods {
-				ds.PodUpdateOrAddIfNotExist(pod)
-			}
-			d := &Director{datastore: ds}
-			gotPod := d.GetRandomPod()
+		period := time.Millisecond
+		factories := []datalayer.EndpointFactory{
+			backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, period),
+			datalayer.NewEndpointFactory([]datalayer.DataSource{&datalayer.FakeDataSource{}}, period),
+		}
+		for _, epf := range factories {
+			t.Run(test.name, func(t *testing.T) {
+				endpointPool := poolutil.InferencePoolToEndpointPool(pool)
+				ds := datastore.NewDatastore(t.Context(), epf, 0)
+				err := ds.PoolSet(t.Context(), fakeClient, endpointPool)
+				if err != nil {
+					t.Errorf("unexpected error setting pool: %s", err)
+				}
+				for _, pod := range test.storePods {
+					ds.PodUpdateOrAddIfNotExist(pod)
+				}
+				d := &Director{datastore: ds}
+				gotPod := d.GetRandomPod()
 
-			if test.expectNil && gotPod != nil {
-				t.Errorf("expected nil pod, got: %v", gotPod)
-			}
-			if !test.expectNil && gotPod == nil {
-				t.Errorf("expected non-nil pod, got nil")
-			}
-		})
+				if test.expectNil && gotPod != nil {
+					t.Errorf("expected nil pod, got: %v", gotPod)
+				}
+				if !test.expectNil && gotPod == nil {
+					t.Errorf("expected non-nil pod, got nil")
+				}
+			})
+		}
 	}
 }