[SPARK-50776][K8S][TESTS] Fix test assertions on executor service acc…

…ount

### What changes were proposed in this pull request?

`ExecutorKubernetesCredentialsFeatureStepSuite` tests that Spark sets the correct service account on executor pods for various configuration combinations. This patch corrects some invalid test assertions and refactors to reduce redundancy across the different test cases.

### Why are the changes needed?

`ExecutorKubernetesCredentialsFeatureStepSuite` attempts to check that the Spark code sets the correct service account on the executor pod. However, the current assertions are actually no-ops that check if a variable is equal to itself, which is always true. The test would pass even if the product code had a bug.

### Does this PR introduce _any_ user-facing change?

No, this is a change in tests only.

### How was this patch tested?

1. Intentionally introduce a bug in `ExecutorKubernetesCredentialsFeatureStep` by hard-coding a bogus service account (not included in this pull request).
1. `build/mvn -o -Pkubernetes -Pscala-2.12 -pl resource-managers/kubernetes/core -Dsuites='org.apache.spark.deploy.k8s.features.ExecutorKubernetesCredentialsFeatureStepSuite' test`
1. The test succeeds, even though there is a bug.
1. Apply this patch.
1. Rerun the test, and it fails, as it should.
1. Revert the bug, rerun the test, and it succeeds.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#49428 from cnauroth/SPARK-50776.

Authored-by: Chris Nauroth <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/ExecutorKubernetesCredentialsFeatureStepSuite.scala

@@ -18,8 +18,10 @@ package org.apache.spark.deploy.k8s.features
 
 import org.scalatest.BeforeAndAfter
 
+import io.fabric8.kubernetes.api.model.PodSpec
+
 import org.apache.spark.{SparkConf, SparkFunSuite}
-import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesTestConf, SparkPod}
+import org.apache.spark.deploy.k8s.{KubernetesTestConf, SparkPod}
 import org.apache.spark.deploy.k8s.Config._
 
 class ExecutorKubernetesCredentialsFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
@@ -30,58 +32,40 @@ class ExecutorKubernetesCredentialsFeatureStepSuite extends SparkFunSuite with B
     baseConf = new SparkConf(false)
   }
 
-  private def newExecutorConf(environment: Map[String, String] = Map.empty):
-  KubernetesExecutorConf = {
-    KubernetesTestConf.createExecutorConf(
-      sparkConf = baseConf,
-      environment = environment)
-  }
-
   test("configure spark pod with executor service account") {
     baseConf.set(KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME, "executor-name")
-    val step = new ExecutorKubernetesCredentialsFeatureStep(newExecutorConf())
-    val spec = step
-      .configurePod(SparkPod.initialPod())
-      .pod
-      .getSpec
-
-    val serviceAccountName = spec.getServiceAccountName
-    val accountName = spec.getServiceAccount
-    assertSAName(serviceAccountName, accountName)
+    val spec = evaluateStep()
+    assertSAName("executor-name", spec)
   }
 
   test("configure spark pod with with driver service account " +
     "and without executor service account") {
     baseConf.set(KUBERNETES_DRIVER_SERVICE_ACCOUNT_NAME, "driver-name")
-    val step = new ExecutorKubernetesCredentialsFeatureStep(newExecutorConf())
-    val spec = step
-      .configurePod(SparkPod.initialPod())
-      .pod
-      .getSpec
-
-    val serviceAccountName = spec.getServiceAccountName
-    val accountName = spec.getServiceAccount
-    assertSAName(serviceAccountName, accountName)
+    val spec = evaluateStep()
+    assertSAName("driver-name", spec)
   }
 
   test("configure spark pod with with driver service account " +
     "and with executor service account") {
     baseConf.set(KUBERNETES_DRIVER_SERVICE_ACCOUNT_NAME, "driver-name")
     baseConf.set(KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME, "executor-name")
+    val spec = evaluateStep()
+    assertSAName("executor-name", spec)
+  }
+
+  private def assertSAName(expectedServiceAccountName: String,
+      spec: PodSpec): Unit = {
+    assert(spec.getServiceAccountName.equals(expectedServiceAccountName))
+    assert(spec.getServiceAccount.equals(expectedServiceAccountName))
+  }
 
-    val step = new ExecutorKubernetesCredentialsFeatureStep(newExecutorConf())
-    val spec = step
+  private def evaluateStep(): PodSpec = {
+    val executorConf = KubernetesTestConf.createExecutorConf(
+        sparkConf = baseConf)
+    val step = new ExecutorKubernetesCredentialsFeatureStep(executorConf)
+    step
       .configurePod(SparkPod.initialPod())
       .pod
       .getSpec
-
-    val serviceAccountName = spec.getServiceAccountName
-    val accountName = spec.getServiceAccount
-    assertSAName(serviceAccountName, accountName)
-  }
-
-  def assertSAName(serviceAccountName: String, accountName: String): Unit = {
-    assert(serviceAccountName.equals(serviceAccountName))
-    assert(accountName.equals(accountName))
   }
 }