Added integration test for pod affinity namespace selector

test/integration/scheduler/predicates_test.go

@@ -26,7 +26,10 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/features"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 	testutils "k8s.io/kubernetes/test/integration/util"
 	imageutils "k8s.io/kubernetes/test/utils/image"
@@ -39,7 +42,7 @@ const pollInterval = 100 * time.Millisecond
 // TestInterPodAffinity verifies that scheduler's inter pod affinity and
 // anti-affinity predicate functions works correctly.
 func TestInterPodAffinity(t *testing.T) {
-	testCtx := initTest(t, "inter-pod-affinity")
+	testCtx := initTest(t, "")
 	defer testutils.CleanupTest(t, testCtx)
 
 	// Add a few nodes with labels
@@ -52,11 +55,18 @@ func TestInterPodAffinity(t *testing.T) {
 	podLabel := map[string]string{"service": "securityscan"}
 	podLabel2 := map[string]string{"security": "S1"}
 
+	if err := createNamespacesWithLabels(cs, []string{"ns1", "ns2"}, map[string]string{"team": "team1"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := createNamespacesWithLabels(cs, []string{"ns3"}, map[string]string{"team": "team2"}); err != nil {
+		t.Fatal(err)
+	}
+	defaultNS := "ns1"
+
 	tests := []struct {
 		name      string
 		pod       *v1.Pod
 		pods      []*v1.Pod
-		node      *v1.Node
 		fits      bool
 		errorType string
 	}{
@@ -88,7 +98,6 @@ func TestInterPodAffinity(t *testing.T) {
 					},
 				},
 			},
-			node:      nodes[0],
 			fits:      false,
 			errorType: "invalidPod",
 		},
@@ -121,7 +130,6 @@ func TestInterPodAffinity(t *testing.T) {
 					},
 				},
 			},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -164,7 +172,6 @@ func TestInterPodAffinity(t *testing.T) {
 				},
 			},
 			},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -202,7 +209,6 @@ func TestInterPodAffinity(t *testing.T) {
 				ObjectMeta: metav1.ObjectMeta{
 					Name:   "fakename2",
 					Labels: podLabel}}},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -240,8 +246,7 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:   "fakename2",
-					Labels: podLabel, Namespace: "ns"}}},
+					Labels: podLabel, Namespace: "ns2"}}},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -278,7 +283,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -332,7 +336,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -386,7 +389,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -439,7 +441,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -516,7 +517,6 @@ func TestInterPodAffinity(t *testing.T) {
 						Labels: podLabel},
 				},
 			},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -569,7 +569,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   nodes[0].Name}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -646,7 +645,6 @@ func TestInterPodAffinity(t *testing.T) {
 						Labels: podLabel},
 				},
 			},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -683,7 +681,6 @@ func TestInterPodAffinity(t *testing.T) {
 				NodeName:   "machine2"}, ObjectMeta: metav1.ObjectMeta{
 				Name:   "fakename2",
 				Labels: podLabel}}},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -723,7 +720,6 @@ func TestInterPodAffinity(t *testing.T) {
 						Labels: podLabel},
 				},
 			},
-			node: nodes[0],
 			fits: false,
 		},
 		{
@@ -763,7 +759,6 @@ func TestInterPodAffinity(t *testing.T) {
 						Labels: podLabel},
 				},
 			},
-			node: nodes[0],
 			fits: true,
 		},
 		{
@@ -805,13 +800,10 @@ func TestInterPodAffinity(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			for _, pod := range test.pods {
-				var nsName string
+				if pod.Namespace == "" {
-				if pod.Namespace != "" {
+					pod.Namespace = defaultNS
-					nsName = pod.Namespace
-				} else {
-					nsName = testCtx.NS.Name
 				}
-				createdPod, err := cs.CoreV1().Pods(nsName).Create(context.TODO(), pod, metav1.CreateOptions{})
+				createdPod, err := cs.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
 				if err != nil {
 					t.Fatalf("Error while creating pod: %v", err)
 				}
@@ -820,7 +812,11 @@ func TestInterPodAffinity(t *testing.T) {
 					t.Errorf("Error while creating pod: %v", err)
 				}
 			}
-			testPod, err := cs.CoreV1().Pods(testCtx.NS.Name).Create(context.TODO(), test.pod, metav1.CreateOptions{})
+			if test.pod.Namespace == "" {
+				test.pod.Namespace = defaultNS
+			}
+
+			testPod, err := cs.CoreV1().Pods(test.pod.Namespace).Create(context.TODO(), test.pod, metav1.CreateOptions{})
 			if err != nil {
 				if !(test.errorType == "invalidPod" && apierrors.IsInvalid(err)) {
 					t.Fatalf("Error while creating pod: %v", err)
@@ -836,7 +832,7 @@ func TestInterPodAffinity(t *testing.T) {
 				t.Errorf("Error while trying to fit a pod: %v", err)
 			}
 
-			err = cs.CoreV1().Pods(testCtx.NS.Name).Delete(context.TODO(), test.pod.Name, *metav1.NewDeleteOptions(0))
+			err = cs.CoreV1().Pods(test.pod.Namespace).Delete(context.TODO(), test.pod.Name, *metav1.NewDeleteOptions(0))
 			if err != nil {
 				t.Errorf("Error while deleting pod: %v", err)
 			}
@@ -845,17 +841,11 @@ func TestInterPodAffinity(t *testing.T) {
 				t.Errorf("Error while waiting for pod to get deleted: %v", err)
 			}
 			for _, pod := range test.pods {
-				var nsName string
+				err = cs.CoreV1().Pods(pod.Namespace).Delete(context.TODO(), pod.Name, *metav1.NewDeleteOptions(0))
-				if pod.Namespace != "" {
-					nsName = pod.Namespace
-				} else {
-					nsName = testCtx.NS.Name
-				}
-				err = cs.CoreV1().Pods(nsName).Delete(context.TODO(), pod.Name, *metav1.NewDeleteOptions(0))
 				if err != nil {
 					t.Errorf("Error while deleting pod: %v", err)
 				}
-				err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodDeleted(cs, nsName, pod.Name))
+				err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodDeleted(cs, pod.Namespace, pod.Name))
 				if err != nil {
 					t.Errorf("Error while waiting for pod to get deleted: %v", err)
 				}
@@ -864,6 +854,241 @@ func TestInterPodAffinity(t *testing.T) {
 	}
 }
 
+// TestInterPodAffinityWithNamespaceSelector verifies that inter pod affinity with NamespaceSelector works as expected.
+// TODO(https://github.com/kubernetes/enhancements/issues/2249): merge with TestInterPodAffinity once NamespaceSelector
+// graduates to GA.
+func TestInterPodAffinityWithNamespaceSelector(t *testing.T) {
+	podLabel := map[string]string{"service": "securityscan"}
+	tests := []struct {
+		name        string
+		pod         *v1.Pod
+		existingPod *v1.Pod
+		fits        bool
+		errorType   string
+		disabled    bool
+	}{
+		{
+			name: "MatchingNamespaces",
+			pod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pod-ns-selector",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+					Affinity: &v1.Affinity{
+						PodAffinity: &v1.PodAffinity{
+							RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
+								{
+									LabelSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "service",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"securityscan"},
+											},
+										},
+									},
+									NamespaceSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "team",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"team1"},
+											},
+										},
+									},
+									TopologyKey: "region",
+								},
+							},
+						},
+					},
+				},
+			},
+			existingPod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "fakename2",
+					Labels:    podLabel,
+					Namespace: "ns2",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+				},
+			},
+			fits: true,
+		},
+		{
+			name: "Disabled",
+			pod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pod-ns-selector",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+					Affinity: &v1.Affinity{
+						PodAffinity: &v1.PodAffinity{
+							RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
+								{
+									LabelSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "service",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"securityscan"},
+											},
+										},
+									},
+									NamespaceSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "team",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"team1"},
+											},
+										},
+									},
+									TopologyKey: "region",
+								},
+							},
+						},
+					},
+				},
+			},
+			existingPod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "fakename2",
+					Labels:    podLabel,
+					Namespace: "ns2",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+				},
+			},
+			fits:     false,
+			disabled: true,
+		},
+		{
+			name: "MismatchingNamespaces",
+			pod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pod-ns-selector",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+					Affinity: &v1.Affinity{
+						PodAffinity: &v1.PodAffinity{
+							RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
+								{
+									LabelSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "service",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"securityscan"},
+											},
+										},
+									},
+									NamespaceSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      "team",
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{"team1"},
+											},
+										},
+									},
+									TopologyKey: "region",
+								},
+							},
+						},
+					},
+				},
+			},
+			existingPod: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "fakename2",
+					Labels:    podLabel,
+					Namespace: "ns3",
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{Name: "container", Image: imageutils.GetPauseImageName()}},
+				},
+			},
+			fits: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.PodAffinityNamespaceSelector, !test.disabled)()
+			testCtx := initTest(t, "")
+			defer testutils.CleanupTest(t, testCtx)
+
+			// Add a few nodes with labels
+			nodes, err := createAndWaitForNodesInCache(testCtx, "testnode", st.MakeNode().Label("region", "r1").Label("zone", "z11"), 2)
+			if err != nil {
+				t.Fatal(err)
+			}
+			test.existingPod.Spec.NodeName = nodes[0].Name
+
+			cs := testCtx.ClientSet
+
+			if err := createNamespacesWithLabels(cs, []string{"ns1", "ns2"}, map[string]string{"team": "team1"}); err != nil {
+				t.Fatal(err)
+			}
+			if err := createNamespacesWithLabels(cs, []string{"ns3"}, map[string]string{"team": "team2"}); err != nil {
+				t.Fatal(err)
+			}
+			defaultNS := "ns1"
+
+			createdPod, err := cs.CoreV1().Pods(test.existingPod.Namespace).Create(context.TODO(), test.existingPod, metav1.CreateOptions{})
+			if err != nil {
+				t.Fatalf("Error while creating pod: %v", err)
+			}
+			err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodScheduled(cs, createdPod.Namespace, createdPod.Name))
+			if err != nil {
+				t.Errorf("Error while creating pod: %v", err)
+			}
+
+			if test.pod.Namespace == "" {
+				test.pod.Namespace = defaultNS
+			}
+
+			testPod, err := cs.CoreV1().Pods(test.pod.Namespace).Create(context.TODO(), test.pod, metav1.CreateOptions{})
+			if err != nil {
+				if !(test.errorType == "invalidPod" && apierrors.IsInvalid(err)) {
+					t.Fatalf("Error while creating pod: %v", err)
+				}
+			}
+
+			if test.fits {
+				err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodScheduled(cs, testPod.Namespace, testPod.Name))
+			} else {
+				err = wait.Poll(pollInterval, wait.ForeverTestTimeout, podUnschedulable(cs, testPod.Namespace, testPod.Name))
+			}
+			if err != nil {
+				t.Errorf("Error while trying to fit a pod: %v", err)
+			}
+
+			err = cs.CoreV1().Pods(test.pod.Namespace).Delete(context.TODO(), test.pod.Name, *metav1.NewDeleteOptions(0))
+			if err != nil {
+				t.Errorf("Error while deleting pod: %v", err)
+			}
+			err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodDeleted(cs, testCtx.NS.Name, test.pod.Name))
+			if err != nil {
+				t.Errorf("Error while waiting for pod to get deleted: %v", err)
+			}
+			err = cs.CoreV1().Pods(test.existingPod.Namespace).Delete(context.TODO(), test.existingPod.Name, *metav1.NewDeleteOptions(0))
+			if err != nil {
+				t.Errorf("Error while deleting pod: %v", err)
+			}
+			err = wait.Poll(pollInterval, wait.ForeverTestTimeout, testutils.PodDeleted(cs, test.existingPod.Namespace, test.existingPod.Name))
+			if err != nil {
+				t.Errorf("Error while waiting for pod to get deleted: %v", err)
+			}
+		})
+	}
+}
+
 // TestEvenPodsSpreadPredicate verifies that EvenPodsSpread predicate functions well.
 func TestEvenPodsSpreadPredicate(t *testing.T) {
 	testCtx := initTest(t, "eps-predicate")

test/integration/scheduler/priorities_test.go

@@ -27,6 +27,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler"
 	schedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/imagelocality"
@@ -119,79 +122,115 @@ func TestNodeAffinity(t *testing.T) {
 // TestPodAffinity verifies that scheduler's pod affinity priority function
 // works correctly.
 func TestPodAffinity(t *testing.T) {
-	testCtx := initTestSchedulerForPriorityTest(t, interpodaffinity.Name)
-	defer testutils.CleanupTest(t, testCtx)
-	// Add a few nodes.
-	topologyKey := "node-topologykey"
-	topologyValue := "topologyvalue"
-	nodesInTopology, err := createAndWaitForNodesInCache(testCtx, "in-topology", st.MakeNode().Label(topologyKey, topologyValue), 5)
-	if err != nil {
-		t.Fatal(err)
-	}
-	// Add a pod with a label and wait for it to schedule.
 	labelKey := "service"
 	labelValue := "S1"
-	_, err = runPausePod(testCtx.ClientSet, initPausePod(&pausePodConfig{
+	topologyKey := "node-topologykey"
-		Name:      "attractor-pod",
+	topologyValue := "topologyvalue"
-		Namespace: testCtx.NS.Name,
+	tests := []struct {
-		Labels:    map[string]string{labelKey: labelValue},
+		name      string
-	}))
+		podConfig *pausePodConfig
-	if err != nil {
+	}{
-		t.Fatalf("Error running the attractor pod: %v", err)
+		{
-	}
+			name: "pod affinity",
-	// Add a few more nodes without the topology label.
+			podConfig: &pausePodConfig{
-	_, err = createAndWaitForNodesInCache(testCtx, "other-node", st.MakeNode(), 5)
+				Name:      "pod1",
-	if err != nil {
+				Namespace: "ns1",
-		t.Fatal(err)
+				Affinity: &v1.Affinity{
-	}
+					PodAffinity: &v1.PodAffinity{
-	// Add a new pod with affinity to the attractor pod.
+						PreferredDuringSchedulingIgnoredDuringExecution: []v1.WeightedPodAffinityTerm{
-	podName := "pod-with-podaffinity"
+							{
-	pod, err := runPausePod(testCtx.ClientSet, initPausePod(&pausePodConfig{
+								PodAffinityTerm: v1.PodAffinityTerm{
-		Name:      podName,
+									LabelSelector: &metav1.LabelSelector{
-		Namespace: testCtx.NS.Name,
+										MatchExpressions: []metav1.LabelSelectorRequirement{
-		Affinity: &v1.Affinity{
+											{
-			PodAffinity: &v1.PodAffinity{
+												Key:      labelKey,
-				PreferredDuringSchedulingIgnoredDuringExecution: []v1.WeightedPodAffinityTerm{
+												Operator: metav1.LabelSelectorOpIn,
-					{
+												Values:   []string{labelValue, "S3"},
-						PodAffinityTerm: v1.PodAffinityTerm{
+											},
-							LabelSelector: &metav1.LabelSelector{
+										},
-								MatchExpressions: []metav1.LabelSelectorRequirement{
-									{
-										Key:      labelKey,
-										Operator: metav1.LabelSelectorOpIn,
-										Values:   []string{labelValue, "S3"},
-									},
-									{
-										Key:      labelKey,
-										Operator: metav1.LabelSelectorOpNotIn,
-										Values:   []string{"S2"},
-									}, {
-										Key:      labelKey,
-										Operator: metav1.LabelSelectorOpExists,
 									},
+									TopologyKey: topologyKey,
 								},
+								Weight: 50,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "pod affinity with namespace selector",
+			podConfig: &pausePodConfig{
+				Name:      "pod1",
+				Namespace: "ns2",
+				Affinity: &v1.Affinity{
+					PodAffinity: &v1.PodAffinity{
+						PreferredDuringSchedulingIgnoredDuringExecution: []v1.WeightedPodAffinityTerm{
+							{
+								PodAffinityTerm: v1.PodAffinityTerm{
+									NamespaceSelector: &metav1.LabelSelector{}, // all namespaces
+									LabelSelector: &metav1.LabelSelector{
+										MatchExpressions: []metav1.LabelSelectorRequirement{
+											{
+												Key:      labelKey,
+												Operator: metav1.LabelSelectorOpIn,
+												Values:   []string{labelValue, "S3"},
+											},
+										},
+									},
+									TopologyKey: topologyKey,
+								},
+								Weight: 50,
 							},
-							TopologyKey: topologyKey,
-							Namespaces:  []string{testCtx.NS.Name},
 						},
-						Weight: 50,
 					},
 				},
 			},
 		},
-	}))
-	if err != nil {
-		t.Fatalf("Error running pause pod: %v", err)
 	}
-	// The new pod must be scheduled on one of the nodes with the same topology
+	for _, tt := range tests {
-	// key-value as the attractor pod.
+		t.Run(tt.name, func(t *testing.T) {
-	for _, node := range nodesInTopology {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.PodAffinityNamespaceSelector, true)()
-		if node.Name == pod.Spec.NodeName {
+			testCtx := initTestSchedulerForPriorityTest(t, interpodaffinity.Name)
-			t.Logf("Pod %v got successfully scheduled on node %v.", podName, pod.Spec.NodeName)
+			defer testutils.CleanupTest(t, testCtx)
-			return
+			// Add a few nodes.
-		}
+			nodesInTopology, err := createAndWaitForNodesInCache(testCtx, "in-topology", st.MakeNode().Label(topologyKey, topologyValue), 5)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := createNamespacesWithLabels(testCtx.ClientSet, []string{"ns1", "ns2"}, map[string]string{"team": "team1"}); err != nil {
+				t.Fatal(err)
+			}
+			// Add a pod with a label and wait for it to schedule.
+			_, err = runPausePod(testCtx.ClientSet, initPausePod(&pausePodConfig{
+				Name:      "attractor-pod",
+				Namespace: "ns1",
+				Labels:    map[string]string{labelKey: labelValue},
+			}))
+			if err != nil {
+				t.Fatalf("Error running the attractor pod: %v", err)
+			}
+			// Add a few more nodes without the topology label.
+			_, err = createAndWaitForNodesInCache(testCtx, "other-node", st.MakeNode(), 5)
+			if err != nil {
+				t.Fatal(err)
+			}
+			// Add a new pod with affinity to the attractor pod.
+			pod, err := runPausePod(testCtx.ClientSet, initPausePod(tt.podConfig))
+			if err != nil {
+				t.Fatalf("Error running pause pod: %v", err)
+			}
+			// The new pod must be scheduled on one of the nodes with the same topology
+			// key-value as the attractor pod.
+			for _, node := range nodesInTopology {
+				if node.Name == pod.Spec.NodeName {
+					t.Logf("Pod %v got successfully scheduled on node %v.", tt.podConfig.Name, pod.Spec.NodeName)
+					return
+				}
+			}
+			t.Errorf("Pod %v got scheduled on an unexpected node: %v.", tt.podConfig.Name, pod.Spec.NodeName)
+		})
 	}
-	t.Errorf("Pod %v got scheduled on an unexpected node: %v.", podName, pod.Spec.NodeName)
 }
 
 // TestImageLocality verifies that the scheduler's image locality priority function

test/integration/scheduler/util.go

@@ -497,3 +497,13 @@ func podScheduled(c clientset.Interface, podNamespace, podName string) wait.Cond
 		return true, nil
 	}
 }
+
+func createNamespacesWithLabels(cs clientset.Interface, namespaces []string, labels map[string]string) error {
+	for _, n := range namespaces {
+		ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: n, Labels: labels}}
+		if _, err := cs.CoreV1().Namespaces().Create(context.TODO(), &ns, metav1.CreateOptions{}); err != nil {
+			return err
+		}
+	}
+	return nil
+}