From 9d7bf4d59738b6a12b0cca90dc2087217e1bf3e4 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Thu, 23 Jul 2026 20:33:03 -0400
Subject: [PATCH 1/6] refactor(computing-unit-managing-service): share CU
 listing helpers and batch k8s calls

Centralize the computing-unit listing logic in ComputingUnitHelpers so the
per-user listing endpoint routes through a single, reusable implementation
(also reused by a follow-up admin endpoint):

- Add resolveOwnerInfo, partitionLiveUnits, reconcileVanishedKubernetesUnits,
  buildDashboardUnit, and the isKubernetes / podPhasesFor / podMetricsFor guards.
- Remove the duplicated private getComputingUnitStatus / getComputingUnitMetrics
  from ComputingUnitManagingResource; callers use the shared helpers.
- Add bulk KubernetesClient.getAllPodPhases / getAllPodMetrics (one namespace-wide
  list()/top() each) plus a shared containerUsage helper and a fetchPodMetricsItems
  extraction; listComputingUnits now resolves pod status/metrics in bulk instead of
  probing podExists twice per unit and re-fetching top() per unit.
- Add a package-private KubernetesClient test seam (setClientForTesting) so the
  fabric8 client can be stubbed in tests; not used in production.

API response behavior is unchanged; per-user reconciliation now runs over
cuid-deduplicated units, reducing redundant DB updates.

Tests: ComputingUnitHelpersSpec (embedded Postgres via MockTexeraDB),
KubernetesClientSpec (stubbed fabric8 client), and a new
ComputingUnitManagingResourceSpec driving the per-user endpoints over the
embedded DB. build.sbt gains DAO/Auth test artifacts and serial test execution.

Generated-by: Claude Code, Claude Opus 4.8
---
 build.sbt                                     |   5 +
 .../ComputingUnitManagingResource.scala       | 140 ++++-------
 .../service/util/ComputingUnitHelpers.scala   | 141 ++++++++++-
 .../service/util/KubernetesClient.scala       |  59 ++++-
 .../ComputingUnitManagingResourceSpec.scala   | 101 ++++++++
 .../util/ComputingUnitHelpersSpec.scala       | 220 +++++++++++++++++-
 .../service/util/KubernetesClientSpec.scala   | 150 +++++++++++-
 7 files changed, 699 insertions(+), 117 deletions(-)
 create mode 100644 computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala

diff --git a/build.sbt b/build.sbt
index ee6882600d8..940b0bcb101 100644
--- a/build.sbt
+++ b/build.sbt
@@ -161,7 +161,12 @@ lazy val WorkflowCore = (project in file("common/workflow-core"))
 lazy val ComputingUnitManagingService = (project in file("computing-unit-managing-service"))
   .dependsOn(WorkflowCore, Auth, Config, Resource)
   .settings(commonModuleSettings)
+  .configs(Test)
+  .dependsOn(DAO % "test->test", Auth % "test->test") // reuse MockTexeraDB embedded Postgres in tests
   .settings(
+    // Tests mutate JVM-wide singletons (SqlServer via MockTexeraDB, the
+    // KubernetesClient fabric8 client), so run suites serially to avoid cross-suite races.
+    Test / parallelExecution := false,
     dependencyOverrides ++= Seq(
       // override it as io.dropwizard 4 require 2.16.1 or higher
       "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion,
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
index aa02f73387e..df67fb3bfe0 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
@@ -50,8 +50,8 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
 import org.apache.texera.service.resource.ComputingUnitManagingResource._
-import org.apache.texera.service.resource.ComputingUnitState._
 import org.apache.texera.service.util.{
+  ComputingUnitHelpers,
   ComputingUnitManagingServiceException,
   InsufficientComputingUnitQuota,
   KubernetesClient
@@ -197,41 +197,6 @@ class ComputingUnitManagingResource {
     }
   }
 
-  private def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = {
-    unit.getType match {
-      // ── Local CUs are always "running" ──────────────────────────────
-      case WorkflowComputingUnitTypeEnum.local =>
-        Running
-
-      // ── Kubernetes CUs – only explicit "Running" counts as running ─
-      case WorkflowComputingUnitTypeEnum.kubernetes =>
-        val phaseOpt = KubernetesClient
-          .getPodByName(KubernetesClient.generatePodName(unit.getCuid))
-          .map(_.getStatus.getPhase)
-
-        if (phaseOpt.contains("Running")) Running else Pending
-
-      // ── Any other (unknown) type is treated as pending ──────────────
-      case _ =>
-        Pending
-    }
-  }
-
-  private def getComputingUnitMetrics(unit: WorkflowComputingUnit): WorkflowComputingUnitMetrics = {
-    unit.getType match {
-      case WorkflowComputingUnitTypeEnum.local =>
-        WorkflowComputingUnitMetrics("NaN", "NaN")
-      case WorkflowComputingUnitTypeEnum.kubernetes =>
-        val metrics = KubernetesClient.getPodMetrics(unit.getCuid)
-        WorkflowComputingUnitMetrics(
-          metrics.getOrElse("cpu", ""),
-          metrics.getOrElse("memory", "")
-        )
-      case _ =>
-        WorkflowComputingUnitMetrics("NaN", "NaN")
-    }
-  }
-
   private def getComputingUnitResourceLimit(
       unit: WorkflowComputingUnit
   ): WorkflowComputingUnitResourceLimit = {
@@ -476,8 +441,8 @@ class ComputingUnitManagingResource {
 
       DashboardWorkflowComputingUnit(
         insertedUnit,
-        getComputingUnitStatus(insertedUnit).toString,
-        getComputingUnitMetrics(insertedUnit),
+        ComputingUnitHelpers.getComputingUnitStatus(insertedUnit).toString,
+        ComputingUnitHelpers.getComputingUnitMetrics(insertedUnit),
         isOwner = true,
         accessPrivilege = PrivilegeEnum.WRITE,
         ownerGoogleAvatar,
@@ -528,63 +493,46 @@ class ComputingUnitManagingResource {
           (List.empty[WorkflowComputingUnit], Map.empty[Integer, PrivilegeEnum])
         }
 
-      val allUnits = ownedUnits ++ sharedUnits
-      val ownerUids: List[Integer] = allUnits.map(_.getUid).distinct
       val userDao = new UserDao(ctx.configuration())
-      val ownerInfoMap: Map[Integer, (String, String)] =
-        userDao
-          .fetchByUid(ownerUids: _*)
-          .asScala
-          .map { u =>
-            val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull
-            val name = Option(u.getName).filter(_.nonEmpty).orNull
-            u.getUid -> (avatar, name)
-          }
-          .toMap
-
-      // If a Kubernetes pod has already disappeared (e.g., manually deleted or TTL
-      // GC-ed by the cluster), we treat the corresponding computing unit as
-      // terminated from the system's point of view. Here we eagerly update its
-      // terminateTime in the database **before** we build the response list so
-      // that subsequent API calls will no longer return this unit.
-      allUnits.foreach { unit =>
-        if (
-          unit.getType == WorkflowComputingUnitTypeEnum.kubernetes &&
-          !KubernetesClient.podExists(unit.getCuid)
-        ) {
-          unit.setTerminateTime(new Timestamp(System.currentTimeMillis()))
-          computingUnitDao.update(unit)
-        }
-      }
 
-      // For shared units, we need to check the access privilege which are saved in different table
-      // to streamline the process, we combine owned units with default WRITE privilege and use sharedUnitInfo
-      // to get the privilege for shared units.
-      (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++ sharedUnits.map(u =>
-        (u, sharedUnitInfo(u.getCuid))
-      ))
-        .distinctBy { case (unit, _) => unit.getCuid }
-        .filter { case (unit, _) => unit.getTerminateTime == null }
-        .filter {
-          case (unit, _) =>
-            unit.getType match {
-              case WorkflowComputingUnitTypeEnum.kubernetes =>
-                KubernetesClient.podExists(unit.getCuid)
-              case _ => true
-            }
-        }
-        .map {
-          case (unit, privilege) =>
-            DashboardWorkflowComputingUnit(
-              computingUnit = unit,
-              isOwner = unit.getUid.equals(uid),
-              accessPrivilege = privilege,
-              status = getComputingUnitStatus(unit).toString,
-              metrics = getComputingUnitMetrics(unit),
-              ownerGoogleAvatar = ownerInfoMap.getOrElse(unit.getUid, (null, null))._1,
-              ownerName = ownerInfoMap.getOrElse(unit.getUid, (null, null))._2
-            )
-        }
+      // Pair each unit with the caller's privilege (owned default to WRITE), one row per cuid, so
+      // a unit that is both owned and shared is reconciled/rendered exactly once.
+      val unitsWithPrivilege =
+        (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++
+          sharedUnits.map(u => (u, sharedUnitInfo(u.getCuid))))
+          .distinctBy { case (unit, _) => unit.getCuid }
+          .filter { case (unit, _) => unit.getTerminateTime == null }
+      val privilegeByCuid = unitsWithPrivilege.map {
+        case (unit, privilege) => unit.getCuid -> privilege
+      }.toMap
+      val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit }
+
+      // Pod phases decide which Kubernetes units are still alive.
+      val podPhases = ComputingUnitHelpers.podPhasesFor(candidateUnits)
+
+      val liveUnits =
+        ComputingUnitHelpers.reconcileVanishedKubernetesUnits(
+          computingUnitDao,
+          candidateUnits,
+          podPhases
+        )
+
+      // Metrics only for survivors, so fetch after reconciliation.
+      val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits)
+
+      val ownerInfoMap =
+        ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct)
+
+      liveUnits.map { unit =>
+        ComputingUnitHelpers.buildDashboardUnit(
+          unit,
+          isOwner = unit.getUid.equals(uid),
+          accessPrivilege = privilegeByCuid(unit.getCuid),
+          ownerInfo = ownerInfoMap,
+          podPhases = podPhases,
+          podMetrics = podMetrics
+        )
+      }
     }
   }
 
@@ -613,8 +561,8 @@ class ComputingUnitManagingResource {
 
     DashboardWorkflowComputingUnit(
       computingUnit = unit,
-      status = getComputingUnitStatus(unit).toString,
-      metrics = getComputingUnitMetrics(unit),
+      status = ComputingUnitHelpers.getComputingUnitStatus(unit).toString,
+      metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit),
       isOwner = unit.getUid.equals(user.getUid),
       accessPrivilege = {
         val cuAccessDao = new ComputingUnitUserAccessDao(context.configuration())
@@ -748,7 +696,7 @@ class ComputingUnitManagingResource {
       throw new BadRequestException("User has no access to the computing unit")
     }
     val computingUnit = getComputingUnitByCuid(context, cuid.toInt)
-    getComputingUnitMetrics(computingUnit)
+    ComputingUnitHelpers.getComputingUnitMetrics(computingUnit)
   }
 
   @GET
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
index 8ae31bba7fb..e877d0209ff 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
@@ -19,11 +19,41 @@
 package org.apache.texera.service.util
 
 import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
+import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao}
 import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
-import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics
+import org.apache.texera.service.resource.ComputingUnitManagingResource.{
+  DashboardWorkflowComputingUnit,
+  WorkflowComputingUnitMetrics
+}
 import org.apache.texera.service.resource.ComputingUnitState.{ComputingUnitState, Pending, Running}
+import org.jooq.EnumType
+
+import java.sql.Timestamp
+import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava}
 
 object ComputingUnitHelpers {
+
+  /**
+    * Owner (avatar, name) keyed by uid, resolved in one query. Blank values collapse to `null`;
+    * empty `uids` returns empty without querying.
+    */
+  def resolveOwnerInfo(
+      userDao: UserDao,
+      uids: Seq[Integer]
+  ): Map[Integer, (String, String)] = {
+    if (uids.isEmpty) Map.empty
+    else
+      userDao
+        .fetchByUid(uids: _*)
+        .asScala
+        .map { u =>
+          val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull
+          val name = Option(u.getName).filter(_.nonEmpty).orNull
+          u.getUid -> (avatar, name)
+        }
+        .toMap
+  }
+
   def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = {
     unit.getType match {
       // Local CUs are always “running”
@@ -58,4 +88,113 @@ object ComputingUnitHelpers {
         WorkflowComputingUnitMetrics("NaN", "NaN")
     }
   }
+
+  /**
+    * Resolves status from a pre-fetched pod-phase map instead of a per-unit cluster call, so a
+    * listing costs O(1) round trips rather than one per unit.
+    */
+  def getComputingUnitStatus(
+      unit: WorkflowComputingUnit,
+      podPhases: Map[String, String]
+  ): ComputingUnitState = {
+    unit.getType match {
+      case WorkflowComputingUnitTypeEnum.local =>
+        Running
+      case WorkflowComputingUnitTypeEnum.kubernetes =>
+        // A missing entry or null phase both count as not-Running.
+        if (podPhases.get(KubernetesClient.generatePodName(unit.getCuid)).contains("Running"))
+          Running
+        else Pending
+      case _ =>
+        Pending
+    }
+  }
+
+  /** Resolves metrics from a pre-fetched pod-metrics map instead of a per-unit cluster call. */
+  def getComputingUnitMetrics(
+      unit: WorkflowComputingUnit,
+      podMetrics: Map[String, Map[String, String]]
+  ): WorkflowComputingUnitMetrics = {
+    unit.getType match {
+      case WorkflowComputingUnitTypeEnum.local =>
+        WorkflowComputingUnitMetrics("NaN", "NaN")
+      case WorkflowComputingUnitTypeEnum.kubernetes =>
+        val metrics = podMetrics
+          .getOrElse(KubernetesClient.generatePodName(unit.getCuid), Map.empty[String, String])
+        WorkflowComputingUnitMetrics(
+          metrics.getOrElse("cpu", ""),
+          metrics.getOrElse("memory", "")
+        )
+      case _ =>
+        WorkflowComputingUnitMetrics("NaN", "NaN")
+    }
+  }
+
+  private def isKubernetes(unit: WorkflowComputingUnit): Boolean =
+    unit.getType match {
+      case WorkflowComputingUnitTypeEnum.kubernetes => true
+      case _                                        => false
+    }
+
+  /** Pod phases for the namespace; skipped (empty) when no Kubernetes unit is present. */
+  def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] =
+    if (units.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty
+
+  /** Pod metrics for the namespace; skipped (empty) when no Kubernetes unit is present. */
+  def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] =
+    if (units.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty
+
+  /** A Kubernetes unit whose pod is absent from `podPhases` (deleted or TTL GC-ed). */
+  private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean =
+    isKubernetes(unit) && !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid))
+
+  /** Partition into `(live, vanished)` by `podPhases`. Pure (no I/O), so it is unit-testable. */
+  def partitionLiveUnits(
+      units: List[WorkflowComputingUnit],
+      podPhases: Map[String, String]
+  ): (List[WorkflowComputingUnit], List[WorkflowComputingUnit]) =
+    units.partition(unit => !isVanished(unit, podPhases))
+
+  /**
+    * Stamp `terminateTime` on vanished Kubernetes units (one batched update) and return the live
+    * ones. Shared by both listing endpoints so they agree on when a unit is terminated.
+    */
+  def reconcileVanishedKubernetesUnits(
+      dao: WorkflowComputingUnitDao,
+      units: List[WorkflowComputingUnit],
+      podPhases: Map[String, String]
+  ): List[WorkflowComputingUnit] = {
+    val partitioned = partitionLiveUnits(units, podPhases)
+    val vanished = partitioned._2
+    if (vanished.nonEmpty) {
+      val now = new Timestamp(System.currentTimeMillis())
+      vanished.foreach(_.setTerminateTime(now))
+      dao.update(vanished.asJava)
+    }
+    partitioned._1
+  }
+
+  /**
+    * Build one dashboard row; status/metrics come from the pre-fetched maps (no per-unit K8s
+    * call). Shared by both listing endpoints so row shape and owner-info fallback stay identical.
+    */
+  def buildDashboardUnit(
+      unit: WorkflowComputingUnit,
+      isOwner: Boolean,
+      accessPrivilege: EnumType,
+      ownerInfo: Map[Integer, (String, String)],
+      podPhases: Map[String, String],
+      podMetrics: Map[String, Map[String, String]]
+  ): DashboardWorkflowComputingUnit = {
+    val owner = ownerInfo.getOrElse(unit.getUid, (null, null))
+    DashboardWorkflowComputingUnit(
+      computingUnit = unit,
+      status = getComputingUnitStatus(unit, podPhases).toString,
+      metrics = getComputingUnitMetrics(unit, podMetrics),
+      isOwner = isOwner,
+      accessPrivilege = accessPrivilege,
+      ownerGoogleAvatar = owner._1,
+      ownerName = owner._2
+    )
+  }
 }
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
index 4f1d391cb30..2f75c3f1562 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
@@ -20,7 +20,7 @@
 package org.apache.texera.service.util
 
 import io.fabric8.kubernetes.api.model._
-import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetricsList
+import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetrics
 import io.fabric8.kubernetes.client.KubernetesClientBuilder
 import org.apache.texera.common.config.KubernetesConfig
 
@@ -28,10 +28,15 @@ import scala.jdk.CollectionConverters._
 
 object KubernetesClient {
 
-  // Initialize the Kubernetes client
-  private val client: io.fabric8.kubernetes.client.KubernetesClient =
+  private var client: io.fabric8.kubernetes.client.KubernetesClient =
     new KubernetesClientBuilder().build()
   private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
+
+  /** Test-only seam to swap in a stubbed client (exercises the wrappers without a live cluster). */
+  private[util] def setClientForTesting(
+      testClient: io.fabric8.kubernetes.client.KubernetesClient
+  ): Unit =
+    client = testClient
   private val podNamePrefix = "computing-unit"
 
   def generatePodURI(cuid: Int): String = {
@@ -48,19 +53,49 @@ object KubernetesClient {
     Option(client.pods().inNamespace(namespace).withName(podName).get())
   }
 
+  /**
+    * Phase of every pod in the namespace, keyed by pod name, in one call — so a bulk listing
+    * avoids a per-unit lookup. Unfiltered so callers can test a unit's presence by its pod-name
+    * key; a pod with no status yet maps to a `null` phase but still appears.
+    */
+  def getAllPodPhases: Map[String, String] = {
+    client
+      .pods()
+      .inNamespace(namespace)
+      .list()
+      .getItems
+      .asScala
+      .map(pod => pod.getMetadata.getName -> Option(pod.getStatus).map(_.getPhase).orNull)
+      .toMap
+  }
+
+  // Flatten a pod's per-container resource usage into a single metric -> value map.
+  private def containerUsage(podMetrics: PodMetrics): Map[String, String] =
+    podMetrics.getContainers.asScala.flatMap { container =>
+      container.getUsage.asScala.map {
+        case (metric, value) => metric -> value.toString
+      }
+    }.toMap
+
+  // One namespace-wide metrics call, returning the raw per-pod items.
+  private def fetchPodMetricsItems(): Iterable[PodMetrics] =
+    client.top().pods().metrics(namespace).getItems.asScala
+
+  /**
+    * CPU/memory of every pod in the namespace, keyed by pod name, in one call — the bulk
+    * counterpart to the single-unit lookup.
+    */
+  def getAllPodMetrics: Map[String, Map[String, String]] =
+    fetchPodMetricsItems()
+      .map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics))
+      .toMap
+
   def getPodMetrics(cuid: Int): Map[String, String] = {
-    val podMetricsList: PodMetricsList = client.top().pods().metrics(namespace)
     val targetPodName = generatePodName(cuid)
-
-    podMetricsList.getItems.asScala
+    fetchPodMetricsItems()
       .collectFirst {
         case podMetrics if podMetrics.getMetadata.getName == targetPodName =>
-          podMetrics.getContainers.asScala.flatMap { container =>
-            container.getUsage.asScala.map {
-              case (metric, value) =>
-                metric -> value.toString
-            }
-          }.toMap
+          containerUsage(podMetrics)
       }
       .getOrElse(Map.empty[String, String])
   }
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
new file mode 100644
index 00000000000..0fb6e214afa
--- /dev/null
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.texera.service.resource
+
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.enums.{
+  PrivilegeEnum,
+  UserRoleEnum,
+  WorkflowComputingUnitTypeEnum
+}
+import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit}
+import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+// Drives the per-user computing-unit endpoints against the embedded database using
+// local units (so no Kubernetes calls are made).
+class ComputingUnitManagingResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with MockTexeraDB {
+
+  private val uid = 800
+  private lazy val user: SessionUser = {
+    val u = new User()
+    u.setUid(uid)
+    u.setName("owner")
+    u.setEmail("owner@example.com")
+    u.setRole(UserRoleEnum.REGULAR)
+    u.setPassword("password")
+    u.setGoogleAvatar("owner-avatar")
+    new SessionUser(u)
+  }
+
+  private def localUnit(cuid: Int, name: String): WorkflowComputingUnit = {
+    val unit = new WorkflowComputingUnit()
+    unit.setCuid(cuid)
+    unit.setUid(uid)
+    unit.setName(name)
+    unit.setType(WorkflowComputingUnitTypeEnum.local)
+    unit
+  }
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    new UserDao(getDSLContext.configuration()).insert(user.getUser)
+    val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration())
+    unitDao.insert(localUnit(800, "cu-a"))
+    unitDao.insert(localUnit(801, "cu-b"))
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  private val resource = new ComputingUnitManagingResource
+
+  "getComputingUnitInfo" should "return the owner's local unit with WRITE access and Running status" in {
+    val info = resource.getComputingUnitInfo(800, user)
+
+    info.computingUnit.getCuid shouldBe 800
+    info.status shouldBe "Running"
+    info.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN")
+    info.isOwner shouldBe true
+    info.accessPrivilege shouldBe PrivilegeEnum.WRITE
+    info.ownerName shouldBe "owner"
+  }
+
+  "getComputingUnitMetricsEndpoint" should "return NaN metrics for an owned local unit" in {
+    resource.getComputingUnitMetricsEndpoint("800", user) shouldBe
+      WorkflowComputingUnitMetrics("NaN", "NaN")
+  }
+
+  "listComputingUnits" should "return the caller's owned, non-terminated units" in {
+    val result = resource.listComputingUnits(user)
+
+    result.map(_.computingUnit.getCuid.intValue()) should contain theSameElementsAs Seq(800, 801)
+    all(result.map(_.isOwner)) shouldBe true
+    all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE
+    all(result.map(_.status)) shouldBe "Running"
+  }
+}
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
index 39a11ea2a54..73deffb4eb2 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
@@ -19,23 +19,63 @@
 
 package org.apache.texera.service.util
 
-import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
-import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.enums.{
+  PrivilegeEnum,
+  UserRoleEnum,
+  WorkflowComputingUnitTypeEnum
+}
+import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit}
 import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics
 import org.apache.texera.service.resource.ComputingUnitState.{Pending, Running}
+import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
-class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers {
+class ComputingUnitHelpersSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with MockTexeraDB {
+
+  override protected def beforeAll(): Unit = initializeDBAndReplaceDSLContext()
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  private lazy val userDao = new UserDao(getDSLContext.configuration())
+  private lazy val computingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration())
 
-  private def localUnit(): WorkflowComputingUnit = {
+  private def makeUser(uid: Int, name: String, email: String, avatar: String): User = {
+    val u = new User()
+    u.setUid(uid)
+    u.setName(name)
+    u.setEmail(email)
+    u.setRole(UserRoleEnum.REGULAR)
+    u.setPassword("password")
+    u.setGoogleAvatar(avatar)
+    u
+  }
+
+  private def makeUnit(
+      cuid: Int,
+      uid: Int,
+      tpe: WorkflowComputingUnitTypeEnum
+  ): WorkflowComputingUnit = {
     val unit = new WorkflowComputingUnit()
-    unit.setType(WorkflowComputingUnitTypeEnum.local)
+    unit.setCuid(cuid)
+    unit.setUid(uid)
+    unit.setType(tpe)
     unit
   }
 
-  // WorkflowComputingUnitTypeEnum only defines `local` and `kubernetes`, so an
-  // untyped unit (getType == null) is what exercises the pure "unknown" branch.
+  private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit =
+    makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.local)
+
+  private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit =
+    makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.kubernetes)
+
+  // A null-type unit (the enum has only local/kubernetes) exercises the "unknown" branch.
   private def untypedUnit(): WorkflowComputingUnit = new WorkflowComputingUnit()
 
   "getComputingUnitStatus" should "return Running for a local unit" in {
@@ -55,4 +95,170 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers {
     ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit()) shouldBe
       WorkflowComputingUnitMetrics("NaN", "NaN")
   }
+
+  // ── Bulk variants resolving from pre-fetched pod maps ────────────────
+
+  "getComputingUnitStatus(unit, podPhases)" should "return Running for a local unit" in {
+    ComputingUnitHelpers.getComputingUnitStatus(localUnit(), Map.empty) shouldBe Running
+  }
+
+  it should "return Running for a kubernetes unit whose pod phase is Running" in {
+    val unit = kubernetesUnit(7)
+    val podPhases = Map(KubernetesClient.generatePodName(7) -> "Running")
+    ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Running
+  }
+
+  it should "return Pending for a kubernetes unit whose pod is absent or not Running" in {
+    val unit = kubernetesUnit(8)
+    ComputingUnitHelpers.getComputingUnitStatus(unit, Map.empty) shouldBe Pending
+    ComputingUnitHelpers.getComputingUnitStatus(
+      unit,
+      Map(KubernetesClient.generatePodName(8) -> "Pending")
+    ) shouldBe Pending
+  }
+
+  it should "treat a null phase as not Running" in {
+    val unit = kubernetesUnit(9)
+    val podPhases = Map(KubernetesClient.generatePodName(9) -> (null: String))
+    ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Pending
+  }
+
+  "getComputingUnitMetrics(unit, podMetrics)" should "return NaN metrics for a local unit" in {
+    ComputingUnitHelpers.getComputingUnitMetrics(localUnit(), Map.empty) shouldBe
+      WorkflowComputingUnitMetrics("NaN", "NaN")
+  }
+
+  it should "resolve cpu/memory for a kubernetes unit from the map" in {
+    val unit = kubernetesUnit(10)
+    val podMetrics = Map(
+      KubernetesClient.generatePodName(10) -> Map("cpu" -> "500m", "memory" -> "256Mi")
+    )
+    ComputingUnitHelpers.getComputingUnitMetrics(unit, podMetrics) shouldBe
+      WorkflowComputingUnitMetrics("500m", "256Mi")
+  }
+
+  it should "return empty cpu/memory for a kubernetes unit absent from the map" in {
+    ComputingUnitHelpers.getComputingUnitMetrics(kubernetesUnit(11), Map.empty) shouldBe
+      WorkflowComputingUnitMetrics("", "")
+  }
+
+  // ── partitionLiveUnits ───────────────────────────────────────────────
+
+  "partitionLiveUnits" should "treat local units as always live" in {
+    val units = List(localUnit(cuid = 1), localUnit(cuid = 2))
+    val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(units, Map.empty)
+    live.map(_.getCuid) shouldBe List(1, 2)
+    vanished shouldBe empty
+  }
+
+  it should "classify a kubernetes unit as live iff its pod is present in the map" in {
+    val present = kubernetesUnit(20)
+    val gone = kubernetesUnit(21)
+    val podPhases = Map(KubernetesClient.generatePodName(20) -> "Running")
+
+    val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(present, gone), podPhases)
+
+    live.map(_.getCuid) shouldBe List(20)
+    vanished.map(_.getCuid) shouldBe List(21)
+  }
+
+  it should "treat an untyped (null-type) unit as live (never kubernetes)" in {
+    val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(untypedUnit()), Map.empty)
+    live should have size 1
+    vanished shouldBe empty
+  }
+
+  // ── buildDashboardUnit ───────────────────────────────────────────────
+
+  "buildDashboardUnit" should "populate the row from the caller flags and pre-fetched maps" in {
+    val unit = kubernetesUnit(cuid = 30, uid = 100)
+    val podName = KubernetesClient.generatePodName(30)
+
+    val row = ComputingUnitHelpers.buildDashboardUnit(
+      unit,
+      isOwner = true,
+      accessPrivilege = PrivilegeEnum.READ,
+      ownerInfo = Map((100: Integer) -> ("avatar", "owner")),
+      podPhases = Map(podName -> "Running"),
+      podMetrics = Map(podName -> Map("cpu" -> "100m", "memory" -> "64Mi"))
+    )
+
+    row.computingUnit.getCuid shouldBe 30
+    row.isOwner shouldBe true
+    row.accessPrivilege shouldBe PrivilegeEnum.READ
+    row.status shouldBe "Running"
+    row.metrics shouldBe WorkflowComputingUnitMetrics("100m", "64Mi")
+    row.ownerGoogleAvatar shouldBe "avatar"
+    row.ownerName shouldBe "owner"
+  }
+
+  it should "fall back to null owner info when the owner is missing from the map" in {
+    val row = ComputingUnitHelpers.buildDashboardUnit(
+      localUnit(cuid = 31, uid = 200),
+      isOwner = false,
+      accessPrivilege = PrivilegeEnum.WRITE,
+      ownerInfo = Map.empty,
+      podPhases = Map.empty,
+      podMetrics = Map.empty
+    )
+
+    row.ownerGoogleAvatar shouldBe null
+    row.ownerName shouldBe null
+    row.status shouldBe "Running"
+    row.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN")
+  }
+
+  // ── Bulk variants: unknown (untyped) branch ──────────────────────────
+
+  "getComputingUnitStatus(unit, podPhases)" should "return Pending for an unknown (untyped) unit" in {
+    ComputingUnitHelpers.getComputingUnitStatus(untypedUnit(), Map.empty) shouldBe Pending
+  }
+
+  "getComputingUnitMetrics(unit, podMetrics)" should "return NaN for an unknown (untyped) unit" in {
+    ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit(), Map.empty) shouldBe
+      WorkflowComputingUnitMetrics("NaN", "NaN")
+  }
+
+  // ── resolveOwnerInfo (backed by the embedded database) ───────────────
+
+  "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in {
+    userDao.insert(makeUser(500, "alice", "alice@example.com", "alice-avatar"))
+    userDao.insert(makeUser(501, "", "bob@example.com", ""))
+
+    val info = ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq[Integer](500, 501))
+    info(500) shouldBe (("alice-avatar", "alice"))
+    info(501) shouldBe ((null, null))
+  }
+
+  it should "return an empty map (and issue no query) for no uids" in {
+    ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq.empty) shouldBe empty
+  }
+
+  // ── reconcileVanishedKubernetesUnits (backed by the embedded database) ─
+
+  "reconcileVanishedKubernetesUnits" should "terminate vanished kubernetes units and return the live ones" in {
+    userDao.insert(makeUser(600, "carol", "carol@example.com", null))
+
+    val present = kubernetesUnit(600, 600)
+    present.setName("present")
+    val gone = kubernetesUnit(601, 600)
+    gone.setName("gone")
+    val local = localUnit(602, 600)
+    local.setName("local")
+    Seq(present, gone, local).foreach(computingUnitDao.insert(_))
+
+    // Only the pod for cuid 600 exists; cuid 601's pod has vanished.
+    val podPhases = Map(KubernetesClient.generatePodName(600) -> "Running")
+    val live =
+      ComputingUnitHelpers.reconcileVanishedKubernetesUnits(
+        computingUnitDao,
+        List(present, gone, local),
+        podPhases
+      )
+
+    live.map(_.getCuid) should contain theSameElementsAs Seq(600, 602)
+    computingUnitDao.fetchOneByCuid(601).getTerminateTime should not be null
+    computingUnitDao.fetchOneByCuid(600).getTerminateTime shouldBe null
+    computingUnitDao.fetchOneByCuid(602).getTerminateTime shouldBe null
+  }
 }
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index 8ce3b107ae4..fc98727e7bc 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -19,10 +19,109 @@
 
 package org.apache.texera.service.util
 
+import io.fabric8.kubernetes.api.model.metrics.v1beta1.{
+  ContainerMetricsBuilder,
+  PodMetricsBuilder,
+  PodMetricsList,
+  PodMetricsListBuilder
+}
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity}
+import io.fabric8.kubernetes.client.dsl.{
+  MetricAPIGroupDSL,
+  MixedOperation,
+  NonNamespaceOperation,
+  PodMetricOperation,
+  PodResource
+}
+import io.fabric8.kubernetes.client.{KubernetesClientBuilder, KubernetesClient => Fabric8Client}
+import org.apache.texera.common.config.KubernetesConfig
+import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
+import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
+import org.mockito.Mockito.{mock, when}
+import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
-class KubernetesClientSpec extends AnyFlatSpec with Matchers {
+import scala.jdk.CollectionConverters._
+
+// Exercises the fabric8 wrappers without a cluster by stubbing the client with Mockito.
+class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll {
+
+  private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
+
+  private def unitOfType(cuid: Int, tpe: WorkflowComputingUnitTypeEnum): WorkflowComputingUnit = {
+    val u = new WorkflowComputingUnit()
+    u.setCuid(cuid)
+    u.setType(tpe)
+    u
+  }
+
+  private def pod(cuid: Int, phase: String): Pod =
+    new PodBuilder()
+      .withNewMetadata()
+      .withName(KubernetesClient.generatePodName(cuid))
+      .endMetadata()
+      .withNewStatus()
+      .withPhase(phase)
+      .endStatus()
+      .build()
+
+  // cuid 1 -> Running, cuid 2 -> Pending; both returned by the namespace-wide listing.
+  private val podList: PodList =
+    new PodListBuilder().addToItems(pod(1, "Running"), pod(2, "Pending")).build()
+
+  // Only cuid 1 reports metrics.
+  private val metricsList: PodMetricsList =
+    new PodMetricsListBuilder()
+      .addToItems(
+        new PodMetricsBuilder()
+          .withNewMetadata()
+          .withName(KubernetesClient.generatePodName(1))
+          .endMetadata()
+          .addToContainers(
+            new ContainerMetricsBuilder()
+              .withName("main")
+              .withUsage(
+                Map("cpu" -> new Quantity("250m"), "memory" -> new Quantity("128Mi")).asJava
+              )
+              .build()
+          )
+          .build()
+      )
+      .build()
+
+  override protected def beforeAll(): Unit = {
+    val client = mock(classOf[Fabric8Client])
+
+    val podsMixed = mock(classOf[MixedOperation[_, _, _]])
+      .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
+    val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])
+      .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]]
+    when(client.pods()).thenReturn(podsMixed)
+    when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace)
+    when(podsInNamespace.list()).thenReturn(podList)
+
+    // cuid 1 -> a present pod; cuid 2 -> no pod.
+    val presentResource = mock(classOf[PodResource])
+    when(presentResource.get()).thenReturn(pod(1, "Running"))
+    val absentResource = mock(classOf[PodResource])
+    when(absentResource.get()).thenReturn(null)
+    when(podsInNamespace.withName(KubernetesClient.generatePodName(1))).thenReturn(presentResource)
+    when(podsInNamespace.withName(KubernetesClient.generatePodName(2))).thenReturn(absentResource)
+
+    val top = mock(classOf[MetricAPIGroupDSL])
+    val podMetricOp = mock(classOf[PodMetricOperation])
+    when(client.top()).thenReturn(top)
+    when(top.pods()).thenReturn(podMetricOp)
+    when(podMetricOp.metrics(namespace)).thenReturn(metricsList)
+
+    KubernetesClient.setClientForTesting(client)
+  }
+
+  override protected def afterAll(): Unit = {
+    // Restore a real client so subsequent suites don't reuse the mock.
+    KubernetesClient.setClientForTesting(new KubernetesClientBuilder().build())
+  }
 
   "generatePodName" should "prefix the cuid with computing-unit" in {
     KubernetesClient.generatePodName(42) shouldBe "computing-unit-42"
@@ -31,4 +130,53 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
   it should "handle a cuid of 0" in {
     KubernetesClient.generatePodName(0) shouldBe "computing-unit-0"
   }
+
+  "getPodByName" should "return the pod when present and None when absent" in {
+    KubernetesClient.getPodByName(KubernetesClient.generatePodName(1)) shouldBe defined
+    KubernetesClient.getPodByName(KubernetesClient.generatePodName(2)) shouldBe None
+  }
+
+  "podExists" should "reflect whether the pod is present" in {
+    KubernetesClient.podExists(1) shouldBe true
+    KubernetesClient.podExists(2) shouldBe false
+  }
+
+  "getAllPodPhases" should "map every pod name in the namespace to its phase" in {
+    val phases = KubernetesClient.getAllPodPhases
+    phases(KubernetesClient.generatePodName(1)) shouldBe "Running"
+    phases(KubernetesClient.generatePodName(2)) shouldBe "Pending"
+  }
+
+  "getAllPodMetrics" should "map pod names to their cpu/memory usage" in {
+    val metrics = KubernetesClient.getAllPodMetrics
+    metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+  }
+
+  "getPodMetrics" should "return the usage for the pod matching the cuid" in {
+    KubernetesClient.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+  }
+
+  it should "return an empty map when no pod matches the cuid" in {
+    KubernetesClient.getPodMetrics(999) shouldBe empty
+  }
+
+  "ComputingUnitHelpers.podPhasesFor" should "query the cluster iff a kubernetes unit is present" in {
+    ComputingUnitHelpers.podPhasesFor(
+      List(unitOfType(5, WorkflowComputingUnitTypeEnum.local))
+    ) shouldBe empty
+
+    ComputingUnitHelpers.podPhasesFor(
+      List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes))
+    )(KubernetesClient.generatePodName(1)) shouldBe "Running"
+  }
+
+  "ComputingUnitHelpers.podMetricsFor" should "query the cluster iff a kubernetes unit is present" in {
+    ComputingUnitHelpers.podMetricsFor(
+      List(unitOfType(5, WorkflowComputingUnitTypeEnum.local))
+    ) shouldBe empty
+
+    ComputingUnitHelpers.podMetricsFor(
+      List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes))
+    )(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+  }
 }

From adcd3775c71f15e718242ef3fb9ba96b1a519bb4 Mon Sep 17 00:00:00 2001
From: "Kunwoo (Chris)" <143021053+kunwp1@users.noreply.github.com>
Date: Thu, 23 Jul 2026 23:28:35 -0400
Subject: [PATCH 2/6] Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Signed-off-by: Kunwoo (Chris) <143021053+kunwp1@users.noreply.github.com>
---
 .../org/apache/texera/service/util/KubernetesClientSpec.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index fc98727e7bc..6bbc3b5c946 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -90,9 +90,9 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter
       )
       .build()
 
-  override protected def beforeAll(): Unit = {
+override protected def beforeAll(): Unit = {
+    super.beforeAll()
     val client = mock(classOf[Fabric8Client])
-
     val podsMixed = mock(classOf[MixedOperation[_, _, _]])
       .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
     val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])

From 195cf166c17d19b25f647ca8a5cfa7e198187e7b Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Thu, 23 Jul 2026 23:38:16 -0400
Subject: [PATCH 3/6] test(computing-unit-managing-service): call super in
 ScalaTest lifecycle hooks

Complete and correct the Copilot-suggested fix on #6853. The auto-pushed commit
only touched KubernetesClientSpec.beforeAll (dedenting the override so it fails
scalafmtCheck) and left the other sites unaddressed.

- Restore correct indentation in KubernetesClientSpec.beforeAll.
- Call super.afterAll() (via try/finally) in KubernetesClientSpec.afterAll.
- Call super.beforeAll()/super.afterAll() in ComputingUnitHelpersSpec and
  ComputingUnitManagingResourceSpec.

So lifecycle code from other mixed-in traits is not skipped if the trait stack
changes; matches the majority convention across the repo's suites. No behavior
change (MockTexeraDB defines no lifecycle hooks). scalafmtCheck clean; the three
specs pass.

Generated-by: Claude Code, Claude Opus 4.8
---
 .../resource/ComputingUnitManagingResourceSpec.scala   |  5 ++++-
 .../texera/service/util/ComputingUnitHelpersSpec.scala |  9 +++++++--
 .../texera/service/util/KubernetesClientSpec.scala     | 10 +++++++---
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
index 0fb6e214afa..54e495473d0 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
@@ -63,6 +63,7 @@ class ComputingUnitManagingResourceSpec
   }
 
   override protected def beforeAll(): Unit = {
+    super.beforeAll()
     initializeDBAndReplaceDSLContext()
     new UserDao(getDSLContext.configuration()).insert(user.getUser)
     val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration())
@@ -70,7 +71,9 @@ class ComputingUnitManagingResourceSpec
     unitDao.insert(localUnit(801, "cu-b"))
   }
 
-  override protected def afterAll(): Unit = shutdownDB()
+  override protected def afterAll(): Unit =
+    try shutdownDB()
+    finally super.afterAll()
 
   private val resource = new ComputingUnitManagingResource
 
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
index 73deffb4eb2..03267f71211 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
@@ -39,9 +39,14 @@ class ComputingUnitHelpersSpec
     with BeforeAndAfterAll
     with MockTexeraDB {
 
-  override protected def beforeAll(): Unit = initializeDBAndReplaceDSLContext()
+  override protected def beforeAll(): Unit = {
+    super.beforeAll()
+    initializeDBAndReplaceDSLContext()
+  }
 
-  override protected def afterAll(): Unit = shutdownDB()
+  override protected def afterAll(): Unit =
+    try shutdownDB()
+    finally super.afterAll()
 
   private lazy val userDao = new UserDao(getDSLContext.configuration())
   private lazy val computingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration())
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index 6bbc3b5c946..aa227a0fb3d 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -90,9 +90,11 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter
       )
       .build()
 
-override protected def beforeAll(): Unit = {
+  override protected def beforeAll(): Unit = {
     super.beforeAll()
+
     val client = mock(classOf[Fabric8Client])
+
     val podsMixed = mock(classOf[MixedOperation[_, _, _]])
       .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
     val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])
@@ -119,8 +121,10 @@ override protected def beforeAll(): Unit = {
   }
 
   override protected def afterAll(): Unit = {
-    // Restore a real client so subsequent suites don't reuse the mock.
-    KubernetesClient.setClientForTesting(new KubernetesClientBuilder().build())
+    try {
+      // Restore a real client so subsequent suites don't reuse the mock.
+      KubernetesClient.setClientForTesting(new KubernetesClientBuilder().build())
+    } finally super.afterAll()
   }
 
   "generatePodName" should "prefix the cuid with computing-unit" in {

From 57dcff735c921ea34fc203b8d5dee54734b34252 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Sat, 25 Jul 2026 09:35:15 -0400
Subject: [PATCH 4/6] refactor(computing-unit-managing-service): drop
 KubernetesClient test var; guard null pod status

Address review feedback on #6853:

- Remove the mutable `var client` / `setClientForTesting` test seam. The
  production singleton is a `val` again; the fabric8 -> map transforms are
  extracted as pure package-private helpers (phasesByPodName, metricsByPodName)
  and tested with builder-constructed model objects, so KubernetesClientSpec
  no longer stubs the client or mutates a JVM-wide global. Drop the low-value
  fabric8 passthrough tests; move the cluster-free podPhasesFor/podMetricsFor
  guard test to ComputingUnitHelpersSpec.
- Guard the single-unit getComputingUnitStatus against a pod that has no
  status yet (Option(getStatus)), matching the bulk getAllPodPhases path, so
  getComputingUnitInfo no longer risks an NPE.
- Update the build.sbt serial-execution note: only MockTexeraDB's SqlServer
  singleton now requires it.

Generated-by: Claude Code, Claude Opus 4.8
---
 build.sbt                                     |   4 +-
 .../service/util/ComputingUnitHelpers.scala   |   4 +-
 .../service/util/KubernetesClient.scala       |  32 ++--
 .../util/ComputingUnitHelpersSpec.scala       |  10 ++
 .../service/util/KubernetesClientSpec.scala   | 158 ++++--------------
 5 files changed, 62 insertions(+), 146 deletions(-)

diff --git a/build.sbt b/build.sbt
index 940b0bcb101..e8f8eb43799 100644
--- a/build.sbt
+++ b/build.sbt
@@ -164,8 +164,8 @@ lazy val ComputingUnitManagingService = (project in file("computing-unit-managin
   .configs(Test)
   .dependsOn(DAO % "test->test", Auth % "test->test") // reuse MockTexeraDB embedded Postgres in tests
   .settings(
-    // Tests mutate JVM-wide singletons (SqlServer via MockTexeraDB, the
-    // KubernetesClient fabric8 client), so run suites serially to avoid cross-suite races.
+    // MockTexeraDB swaps a JVM-wide singleton (SqlServer's embedded Postgres),
+    // so run suites serially to avoid cross-suite races.
     Test / parallelExecution := false,
     dependencyOverrides ++= Seq(
       // override it as io.dropwizard 4 require 2.16.1 or higher
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
index e877d0209ff..1fdbb664fcb 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
@@ -62,9 +62,11 @@ object ComputingUnitHelpers {
 
       // Kubernetes CUs – only explicit “Running” counts as running
       case WorkflowComputingUnitTypeEnum.kubernetes =>
+        // Guard the pod status the same way the bulk getAllPodPhases does: a pod with no
+        // status yet has a null getStatus, so map through Option to avoid an NPE.
         val phaseOpt = KubernetesClient
           .getPodByName(KubernetesClient.generatePodName(unit.getCuid))
-          .map(_.getStatus.getPhase)
+          .flatMap(pod => Option(pod.getStatus).map(_.getPhase))
 
         if (phaseOpt.contains("Running")) Running else Pending
 
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
index 2f75c3f1562..8b99822f79b 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
@@ -28,15 +28,9 @@ import scala.jdk.CollectionConverters._
 
 object KubernetesClient {
 
-  private var client: io.fabric8.kubernetes.client.KubernetesClient =
+  private val client: io.fabric8.kubernetes.client.KubernetesClient =
     new KubernetesClientBuilder().build()
   private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
-
-  /** Test-only seam to swap in a stubbed client (exercises the wrappers without a live cluster). */
-  private[util] def setClientForTesting(
-      testClient: io.fabric8.kubernetes.client.KubernetesClient
-  ): Unit =
-    client = testClient
   private val podNamePrefix = "computing-unit"
 
   def generatePodURI(cuid: Int): String = {
@@ -58,16 +52,14 @@ object KubernetesClient {
     * avoids a per-unit lookup. Unfiltered so callers can test a unit's presence by its pod-name
     * key; a pod with no status yet maps to a `null` phase but still appears.
     */
-  def getAllPodPhases: Map[String, String] = {
-    client
-      .pods()
-      .inNamespace(namespace)
-      .list()
-      .getItems
-      .asScala
+  def getAllPodPhases: Map[String, String] =
+    phasesByPodName(client.pods().inNamespace(namespace).list().getItems.asScala)
+
+  /** Pure fabric8 -> map transform: a pod with no status yet maps to a `null` phase. */
+  private[util] def phasesByPodName(pods: Iterable[Pod]): Map[String, String] =
+    pods
       .map(pod => pod.getMetadata.getName -> Option(pod.getStatus).map(_.getPhase).orNull)
       .toMap
-  }
 
   // Flatten a pod's per-container resource usage into a single metric -> value map.
   private def containerUsage(podMetrics: PodMetrics): Map[String, String] =
@@ -77,6 +69,12 @@ object KubernetesClient {
       }
     }.toMap
 
+  /** Pure fabric8 -> map transform over the raw per-pod metrics items. */
+  private[util] def metricsByPodName(
+      items: Iterable[PodMetrics]
+  ): Map[String, Map[String, String]] =
+    items.map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics)).toMap
+
   // One namespace-wide metrics call, returning the raw per-pod items.
   private def fetchPodMetricsItems(): Iterable[PodMetrics] =
     client.top().pods().metrics(namespace).getItems.asScala
@@ -86,9 +84,7 @@ object KubernetesClient {
     * counterpart to the single-unit lookup.
     */
   def getAllPodMetrics: Map[String, Map[String, String]] =
-    fetchPodMetricsItems()
-      .map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics))
-      .toMap
+    metricsByPodName(fetchPodMetricsItems())
 
   def getPodMetrics(cuid: Int): Map[String, String] = {
     val targetPodName = generatePodName(cuid)
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
index 03267f71211..8d30d99e7e0 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
@@ -224,6 +224,16 @@ class ComputingUnitHelpersSpec
       WorkflowComputingUnitMetrics("NaN", "NaN")
   }
 
+  // ── podPhasesFor / podMetricsFor guards ──────────────────────────────
+
+  "podPhasesFor" should "return empty (issuing no cluster call) when no kubernetes unit is present" in {
+    ComputingUnitHelpers.podPhasesFor(List(localUnit(), untypedUnit())) shouldBe empty
+  }
+
+  "podMetricsFor" should "return empty (issuing no cluster call) when no kubernetes unit is present" in {
+    ComputingUnitHelpers.podMetricsFor(List(localUnit(), untypedUnit())) shouldBe empty
+  }
+
   // ── resolveOwnerInfo (backed by the embedded database) ───────────────
 
   "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in {
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index aa227a0fb3d..ce4cc9b850f 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -21,40 +21,19 @@ package org.apache.texera.service.util
 
 import io.fabric8.kubernetes.api.model.metrics.v1beta1.{
   ContainerMetricsBuilder,
-  PodMetricsBuilder,
-  PodMetricsList,
-  PodMetricsListBuilder
+  PodMetrics,
+  PodMetricsBuilder
 }
-import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity}
-import io.fabric8.kubernetes.client.dsl.{
-  MetricAPIGroupDSL,
-  MixedOperation,
-  NonNamespaceOperation,
-  PodMetricOperation,
-  PodResource
-}
-import io.fabric8.kubernetes.client.{KubernetesClientBuilder, KubernetesClient => Fabric8Client}
-import org.apache.texera.common.config.KubernetesConfig
-import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
-import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
-import org.mockito.Mockito.{mock, when}
-import org.scalatest.BeforeAndAfterAll
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, Quantity}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import scala.jdk.CollectionConverters._
 
-// Exercises the fabric8 wrappers without a cluster by stubbing the client with Mockito.
-class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll {
-
-  private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
-
-  private def unitOfType(cuid: Int, tpe: WorkflowComputingUnitTypeEnum): WorkflowComputingUnit = {
-    val u = new WorkflowComputingUnit()
-    u.setCuid(cuid)
-    u.setType(tpe)
-    u
-  }
+// Exercises the pure fabric8 -> map transforms with builder-constructed model objects, so no
+// cluster or client stubbing is needed. The status/metrics *decision* logic that consumes these
+// maps (Running vs Pending, cpu/memory resolution) is covered by ComputingUnitHelpersSpec.
+class KubernetesClientSpec extends AnyFlatSpec with Matchers {
 
   private def pod(cuid: Int, phase: String): Pod =
     new PodBuilder()
@@ -66,67 +45,27 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter
       .endStatus()
       .build()
 
-  // cuid 1 -> Running, cuid 2 -> Pending; both returned by the namespace-wide listing.
-  private val podList: PodList =
-    new PodListBuilder().addToItems(pod(1, "Running"), pod(2, "Pending")).build()
+  // A pod whose status has not been populated yet (getStatus == null).
+  private def statuslessPod(cuid: Int): Pod =
+    new PodBuilder()
+      .withNewMetadata()
+      .withName(KubernetesClient.generatePodName(cuid))
+      .endMetadata()
+      .build()
 
-  // Only cuid 1 reports metrics.
-  private val metricsList: PodMetricsList =
-    new PodMetricsListBuilder()
-      .addToItems(
-        new PodMetricsBuilder()
-          .withNewMetadata()
-          .withName(KubernetesClient.generatePodName(1))
-          .endMetadata()
-          .addToContainers(
-            new ContainerMetricsBuilder()
-              .withName("main")
-              .withUsage(
-                Map("cpu" -> new Quantity("250m"), "memory" -> new Quantity("128Mi")).asJava
-              )
-              .build()
-          )
+  private def podMetrics(cuid: Int, cpu: String, memory: String): PodMetrics =
+    new PodMetricsBuilder()
+      .withNewMetadata()
+      .withName(KubernetesClient.generatePodName(cuid))
+      .endMetadata()
+      .addToContainers(
+        new ContainerMetricsBuilder()
+          .withName("main")
+          .withUsage(Map("cpu" -> new Quantity(cpu), "memory" -> new Quantity(memory)).asJava)
           .build()
       )
       .build()
 
-  override protected def beforeAll(): Unit = {
-    super.beforeAll()
-
-    val client = mock(classOf[Fabric8Client])
-
-    val podsMixed = mock(classOf[MixedOperation[_, _, _]])
-      .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
-    val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])
-      .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]]
-    when(client.pods()).thenReturn(podsMixed)
-    when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace)
-    when(podsInNamespace.list()).thenReturn(podList)
-
-    // cuid 1 -> a present pod; cuid 2 -> no pod.
-    val presentResource = mock(classOf[PodResource])
-    when(presentResource.get()).thenReturn(pod(1, "Running"))
-    val absentResource = mock(classOf[PodResource])
-    when(absentResource.get()).thenReturn(null)
-    when(podsInNamespace.withName(KubernetesClient.generatePodName(1))).thenReturn(presentResource)
-    when(podsInNamespace.withName(KubernetesClient.generatePodName(2))).thenReturn(absentResource)
-
-    val top = mock(classOf[MetricAPIGroupDSL])
-    val podMetricOp = mock(classOf[PodMetricOperation])
-    when(client.top()).thenReturn(top)
-    when(top.pods()).thenReturn(podMetricOp)
-    when(podMetricOp.metrics(namespace)).thenReturn(metricsList)
-
-    KubernetesClient.setClientForTesting(client)
-  }
-
-  override protected def afterAll(): Unit = {
-    try {
-      // Restore a real client so subsequent suites don't reuse the mock.
-      KubernetesClient.setClientForTesting(new KubernetesClientBuilder().build())
-    } finally super.afterAll()
-  }
-
   "generatePodName" should "prefix the cuid with computing-unit" in {
     KubernetesClient.generatePodName(42) shouldBe "computing-unit-42"
   }
@@ -135,52 +74,21 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter
     KubernetesClient.generatePodName(0) shouldBe "computing-unit-0"
   }
 
-  "getPodByName" should "return the pod when present and None when absent" in {
-    KubernetesClient.getPodByName(KubernetesClient.generatePodName(1)) shouldBe defined
-    KubernetesClient.getPodByName(KubernetesClient.generatePodName(2)) shouldBe None
-  }
-
-  "podExists" should "reflect whether the pod is present" in {
-    KubernetesClient.podExists(1) shouldBe true
-    KubernetesClient.podExists(2) shouldBe false
-  }
-
-  "getAllPodPhases" should "map every pod name in the namespace to its phase" in {
-    val phases = KubernetesClient.getAllPodPhases
+  "phasesByPodName" should "map every pod name to its phase" in {
+    val phases = KubernetesClient.phasesByPodName(Seq(pod(1, "Running"), pod(2, "Pending")))
     phases(KubernetesClient.generatePodName(1)) shouldBe "Running"
     phases(KubernetesClient.generatePodName(2)) shouldBe "Pending"
   }
 
-  "getAllPodMetrics" should "map pod names to their cpu/memory usage" in {
-    val metrics = KubernetesClient.getAllPodMetrics
-    metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+  it should "map a pod with no status to a null phase but still include it" in {
+    val phases = KubernetesClient.phasesByPodName(Seq(statuslessPod(3)))
+    phases should contain key KubernetesClient.generatePodName(3)
+    phases(KubernetesClient.generatePodName(3)) shouldBe null
   }
 
-  "getPodMetrics" should "return the usage for the pod matching the cuid" in {
-    KubernetesClient.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
-  }
-
-  it should "return an empty map when no pod matches the cuid" in {
-    KubernetesClient.getPodMetrics(999) shouldBe empty
-  }
-
-  "ComputingUnitHelpers.podPhasesFor" should "query the cluster iff a kubernetes unit is present" in {
-    ComputingUnitHelpers.podPhasesFor(
-      List(unitOfType(5, WorkflowComputingUnitTypeEnum.local))
-    ) shouldBe empty
-
-    ComputingUnitHelpers.podPhasesFor(
-      List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes))
-    )(KubernetesClient.generatePodName(1)) shouldBe "Running"
-  }
-
-  "ComputingUnitHelpers.podMetricsFor" should "query the cluster iff a kubernetes unit is present" in {
-    ComputingUnitHelpers.podMetricsFor(
-      List(unitOfType(5, WorkflowComputingUnitTypeEnum.local))
-    ) shouldBe empty
-
-    ComputingUnitHelpers.podMetricsFor(
-      List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes))
-    )(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+  "metricsByPodName" should "flatten each pod's container usage into a cpu/memory map" in {
+    val metrics =
+      KubernetesClient.metricsByPodName(Seq(podMetrics(1, "250m", "128Mi")))
+    metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
   }
 }

From 91cbdb458d7ee898da218314b96aab94322ea1c0 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Sun, 26 Jul 2026 01:52:31 -0400
Subject: [PATCH 5/6] test(computing-unit-managing-service): cover k8s
 passthrough wrappers via injected client

Reach 100% patch coverage on the CU listing refactor by making the fabric8
passthrough wrappers testable without the mutable-global seam removed earlier
in review:

- KubernetesClient becomes `class KubernetesClient(client)` plus a companion
  `object` bound to the real in-cluster client, so tests construct an instance
  over a stubbed fabric8 client. No mutable global, no setClientForTesting; all
  KubernetesClient.<member> call sites are unchanged (inherited on the object).
- ComputingUnitHelpers threads the client through private[util] by-name
  overloads (singleUnitStatus, podPhasesFor, podMetricsFor): the local/empty
  path never forces the singleton, and the kubernetes branches become
  unit-testable with a stub. Public signatures and production call sites are
  unchanged; only the modified null-guard path gets a seam.
- KubernetesClientSpec adds wrapper tests over a stubbed fabric8 fluent chain;
  ComputingUnitHelpersSpec adds kubernetes-branch tests with a mocked
  KubernetesClient at the consumer boundary.
---
 .../service/util/ComputingUnitHelpers.scala   | 44 +++++++++--
 .../service/util/KubernetesClient.scala       | 13 +++-
 .../util/ComputingUnitHelpersSpec.scala       | 45 +++++++++++
 .../service/util/KubernetesClientSpec.scala   | 78 +++++++++++++++++--
 4 files changed, 165 insertions(+), 15 deletions(-)

diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
index 1fdbb664fcb..ea7a65ef517 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
@@ -54,7 +54,21 @@ object ComputingUnitHelpers {
         .toMap
   }
 
-  def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = {
+  def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState =
+    singleUnitStatus(unit, KubernetesClient)
+
+  /**
+    * Single-unit status via a per-unit pod lookup (a targeted GET, cheaper than listing the whole
+    * namespace). The client is a by-name parameter — not the global singleton — so the kubernetes
+    * branch is unit-testable with a stub and the local/unknown branches never force the singleton;
+    * the public overload binds the production [[KubernetesClient]]. (Metrics has no analogous seam:
+    * its per-unit lookup already fans out to the whole namespace and the bulk (unit, podMetrics)
+    * overload already covers the cpu/memory resolution, so nothing there is worth pinning.)
+    */
+  private[util] def singleUnitStatus(
+      unit: WorkflowComputingUnit,
+      k8s: => KubernetesClient
+  ): ComputingUnitState = {
     unit.getType match {
       // Local CUs are always “running”
       case WorkflowComputingUnitTypeEnum.local =>
@@ -64,8 +78,9 @@ object ComputingUnitHelpers {
       case WorkflowComputingUnitTypeEnum.kubernetes =>
         // Guard the pod status the same way the bulk getAllPodPhases does: a pod with no
         // status yet has a null getStatus, so map through Option to avoid an NPE.
-        val phaseOpt = KubernetesClient
-          .getPodByName(KubernetesClient.generatePodName(unit.getCuid))
+        val client = k8s
+        val phaseOpt = client
+          .getPodByName(client.generatePodName(unit.getCuid))
           .flatMap(pod => Option(pod.getStatus).map(_.getPhase))
 
         if (phaseOpt.contains("Running")) Running else Pending
@@ -138,13 +153,28 @@ object ComputingUnitHelpers {
       case _                                        => false
     }
 
-  /** Pod phases for the namespace; skipped (empty) when no Kubernetes unit is present. */
+  // Pod phases/metrics for the namespace; skipped (empty) when no Kubernetes unit is present, so a
+  // cluster-free listing issues no round trip. Same seam as singleUnitStatus: the public overload
+  // binds the production singleton, passing it by-name to the private[util] overload, which forces
+  // it only inside the guard's true branch — so the empty path never touches the client, and tests
+  // drive the private overload with a stub.
   def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] =
-    if (units.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty
+    podPhasesFor(units, KubernetesClient)
+
+  private[util] def podPhasesFor(
+      units: Seq[WorkflowComputingUnit],
+      k8s: => KubernetesClient
+  ): Map[String, String] =
+    if (units.exists(isKubernetes)) k8s.getAllPodPhases else Map.empty
 
-  /** Pod metrics for the namespace; skipped (empty) when no Kubernetes unit is present. */
   def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] =
-    if (units.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty
+    podMetricsFor(units, KubernetesClient)
+
+  private[util] def podMetricsFor(
+      units: Seq[WorkflowComputingUnit],
+      k8s: => KubernetesClient
+  ): Map[String, Map[String, String]] =
+    if (units.exists(isKubernetes)) k8s.getAllPodMetrics else Map.empty
 
   /** A Kubernetes unit whose pod is absent from `podPhases` (deleted or TTL GC-ed). */
   private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean =
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
index 8b99822f79b..6f97a5cf6b1 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
@@ -26,10 +26,14 @@ import org.apache.texera.common.config.KubernetesConfig
 
 import scala.jdk.CollectionConverters._
 
-object KubernetesClient {
+/**
+  * Thin wrapper over the fabric8 Kubernetes client. The production singleton is the companion
+  * object below, bound to a real in-cluster client. The fabric8 client is a constructor
+  * parameter (not a mutable global) so tests can construct an instance backed by a stubbed
+  * client and exercise the passthrough wrappers without a live cluster.
+  */
+class KubernetesClient(client: io.fabric8.kubernetes.client.KubernetesClient) {
 
-  private val client: io.fabric8.kubernetes.client.KubernetesClient =
-    new KubernetesClientBuilder().build()
   private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
   private val podNamePrefix = "computing-unit"
 
@@ -213,3 +217,6 @@ object KubernetesClient {
     client.pods().inNamespace(namespace).withName(generatePodName(cuid)).delete()
   }
 }
+
+/** Production singleton bound to a real in-cluster fabric8 client. */
+object KubernetesClient extends KubernetesClient(new KubernetesClientBuilder().build())
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
index 8d30d99e7e0..cf54089a132 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.texera.service.util
 
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder}
 import org.apache.texera.dao.MockTexeraDB
 import org.apache.texera.dao.jooq.generated.enums.{
   PrivilegeEnum,
@@ -29,6 +30,7 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComput
 import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit}
 import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics
 import org.apache.texera.service.resource.ComputingUnitState.{Pending, Running}
+import org.mockito.Mockito.{mock, when}
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -83,6 +85,12 @@ class ComputingUnitHelpersSpec
   // A null-type unit (the enum has only local/kubernetes) exercises the "unknown" branch.
   private def untypedUnit(): WorkflowComputingUnit = new WorkflowComputingUnit()
 
+  private def podWithPhase(phase: String): Pod =
+    new PodBuilder().withNewStatus().withPhase(phase).endStatus().build()
+
+  // A pod whose status has not been populated yet (getStatus == null).
+  private def statuslessPod(): Pod = new PodBuilder().build()
+
   "getComputingUnitStatus" should "return Running for a local unit" in {
     ComputingUnitHelpers.getComputingUnitStatus(localUnit()) shouldBe Running
   }
@@ -91,6 +99,29 @@ class ComputingUnitHelpersSpec
     ComputingUnitHelpers.getComputingUnitStatus(untypedUnit()) shouldBe Pending
   }
 
+  // The kubernetes branch does a per-unit pod lookup, so singleUnitStatus is driven through a
+  // stubbed client (the public getComputingUnitStatus binds the production singleton).
+  "singleUnitStatus" should "return Running for a kubernetes unit whose pod phase is Running" in {
+    val k8s = mock(classOf[KubernetesClient])
+    when(k8s.generatePodName(40)).thenReturn("computing-unit-40")
+    when(k8s.getPodByName("computing-unit-40")).thenReturn(Some(podWithPhase("Running")))
+    ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(40), k8s) shouldBe Running
+  }
+
+  it should "return Pending for a kubernetes unit whose pod has no status yet" in {
+    val k8s = mock(classOf[KubernetesClient])
+    when(k8s.generatePodName(41)).thenReturn("computing-unit-41")
+    when(k8s.getPodByName("computing-unit-41")).thenReturn(Some(statuslessPod()))
+    ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(41), k8s) shouldBe Pending
+  }
+
+  it should "return Pending for a kubernetes unit whose pod is absent" in {
+    val k8s = mock(classOf[KubernetesClient])
+    when(k8s.generatePodName(42)).thenReturn("computing-unit-42")
+    when(k8s.getPodByName("computing-unit-42")).thenReturn(None)
+    ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(42), k8s) shouldBe Pending
+  }
+
   "getComputingUnitMetrics" should "return NaN metrics for a local unit" in {
     ComputingUnitHelpers.getComputingUnitMetrics(localUnit()) shouldBe
       WorkflowComputingUnitMetrics("NaN", "NaN")
@@ -230,10 +261,24 @@ class ComputingUnitHelpersSpec
     ComputingUnitHelpers.podPhasesFor(List(localUnit(), untypedUnit())) shouldBe empty
   }
 
+  it should "fetch all pod phases once when a kubernetes unit is present" in {
+    val k8s = mock(classOf[KubernetesClient])
+    val phases = Map("computing-unit-50" -> "Running")
+    when(k8s.getAllPodPhases).thenReturn(phases)
+    ComputingUnitHelpers.podPhasesFor(List(kubernetesUnit(50)), k8s) shouldBe phases
+  }
+
   "podMetricsFor" should "return empty (issuing no cluster call) when no kubernetes unit is present" in {
     ComputingUnitHelpers.podMetricsFor(List(localUnit(), untypedUnit())) shouldBe empty
   }
 
+  it should "fetch all pod metrics once when a kubernetes unit is present" in {
+    val k8s = mock(classOf[KubernetesClient])
+    val metrics = Map("computing-unit-51" -> Map("cpu" -> "100m", "memory" -> "64Mi"))
+    when(k8s.getAllPodMetrics).thenReturn(metrics)
+    ComputingUnitHelpers.podMetricsFor(List(kubernetesUnit(51)), k8s) shouldBe metrics
+  }
+
   // ── resolveOwnerInfo (backed by the embedded database) ───────────────
 
   "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in {
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index ce4cc9b850f..f37969f0173 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -22,19 +22,63 @@ package org.apache.texera.service.util
 import io.fabric8.kubernetes.api.model.metrics.v1beta1.{
   ContainerMetricsBuilder,
   PodMetrics,
-  PodMetricsBuilder
+  PodMetricsBuilder,
+  PodMetricsList,
+  PodMetricsListBuilder
 }
-import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, Quantity}
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity}
+import io.fabric8.kubernetes.client.dsl.{
+  MetricAPIGroupDSL,
+  MixedOperation,
+  NonNamespaceOperation,
+  PodMetricOperation,
+  PodResource
+}
+import io.fabric8.kubernetes.client.{KubernetesClient => Fabric8Client}
+import org.apache.texera.common.config.KubernetesConfig
+import org.mockito.Mockito.{mock, when}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import scala.jdk.CollectionConverters._
 
-// Exercises the pure fabric8 -> map transforms with builder-constructed model objects, so no
-// cluster or client stubbing is needed. The status/metrics *decision* logic that consumes these
-// maps (Running vs Pending, cpu/memory resolution) is covered by ComputingUnitHelpersSpec.
+// Two layers are exercised here:
+//   * the pure fabric8 -> map transforms (phasesByPodName / metricsByPodName) with
+//     builder-constructed model objects, so the transform logic needs no client, and
+//   * the thin namespace-wide wrappers (getAllPodPhases / getAllPodMetrics / getPodMetrics),
+//     which are driven through a freshly constructed KubernetesClient whose fabric8 client is a
+//     Mockito stub — no live cluster and no mutable global.
+// The status/metrics *decision* logic that consumes these maps (Running vs Pending, cpu/memory
+// resolution) is covered by ComputingUnitHelpersSpec.
 class KubernetesClientSpec extends AnyFlatSpec with Matchers {
 
+  private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
+
+  // A fabric8 client stubbed just enough to answer the namespace-wide pod-list and pod-metrics
+  // calls the wrappers make. RETURNS_DEEP_STUBS can't be used: fabric8's fluent API returns type
+  // variables, so each step of the chain is mocked explicitly.
+  private def stubbedClient(pods: Seq[Pod], metrics: Seq[PodMetrics]): Fabric8Client = {
+    val client = mock(classOf[Fabric8Client])
+
+    val podsMixed = mock(classOf[MixedOperation[_, _, _]])
+      .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
+    val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])
+      .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]]
+    when(client.pods()).thenReturn(podsMixed)
+    when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace)
+    when(podsInNamespace.list()).thenReturn(new PodListBuilder().addAllToItems(pods.asJava).build())
+
+    val top = mock(classOf[MetricAPIGroupDSL])
+    val podMetricOp = mock(classOf[PodMetricOperation])
+    val metricsList: PodMetricsList =
+      new PodMetricsListBuilder().addAllToItems(metrics.asJava).build()
+    when(client.top()).thenReturn(top)
+    when(top.pods()).thenReturn(podMetricOp)
+    when(podMetricOp.metrics(namespace)).thenReturn(metricsList)
+
+    client
+  }
+
   private def pod(cuid: Int, phase: String): Pod =
     new PodBuilder()
       .withNewMetadata()
@@ -91,4 +135,28 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
       KubernetesClient.metricsByPodName(Seq(podMetrics(1, "250m", "128Mi")))
     metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
   }
+
+  // ── namespace-wide wrappers, driven through a stubbed fabric8 client ──
+  // These pin the fabric8 fluent-chain plumbing (list() / top().pods().metrics()); the value
+  // transform they delegate to is already pinned by the phasesByPodName / metricsByPodName tests,
+  // so they only assert that the namespace items flow through keyed by pod name.
+
+  "getAllPodPhases" should "list the namespace pods and key them by pod name" in {
+    val k8s = new KubernetesClient(stubbedClient(Seq(pod(1, "Running"), pod(2, "Pending")), Seq.empty))
+    k8s.getAllPodPhases.keySet shouldBe
+      Set(KubernetesClient.generatePodName(1), KubernetesClient.generatePodName(2))
+  }
+
+  "getAllPodMetrics" should "fetch the namespace metrics and key them by pod name" in {
+    val k8s = new KubernetesClient(stubbedClient(Seq.empty, Seq(podMetrics(1, "250m", "128Mi"))))
+    k8s.getAllPodMetrics.keySet shouldBe Set(KubernetesClient.generatePodName(1))
+  }
+
+  // getPodMetrics adds its own collectFirst-by-name lookup on top of the transform, so it asserts
+  // both the matched pod's usage and the no-match fallback.
+  "getPodMetrics" should "return the matching pod's usage and an empty map when none matches" in {
+    val k8s = new KubernetesClient(stubbedClient(Seq.empty, Seq(podMetrics(1, "250m", "128Mi"))))
+    k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
+    k8s.getPodMetrics(999) shouldBe empty
+  }
 }

From f6fca1d4e6dfc15c3648127a6b9ad768a5d93b06 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Sun, 26 Jul 2026 02:04:16 -0400
Subject: [PATCH 6/6] style(computing-unit-managing-service): scalafmt
 KubernetesClientSpec

Wrap the over-long new KubernetesClient(...) line in the getAllPodPhases test
to satisfy scalafmtCheck. Formatting only.
---
 .../org/apache/texera/service/util/KubernetesClientSpec.scala  | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index f37969f0173..f0d96347a91 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -142,7 +142,8 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
   // so they only assert that the namespace items flow through keyed by pod name.
 
   "getAllPodPhases" should "list the namespace pods and key them by pod name" in {
-    val k8s = new KubernetesClient(stubbedClient(Seq(pod(1, "Running"), pod(2, "Pending")), Seq.empty))
+    val k8s =
+      new KubernetesClient(stubbedClient(Seq(pod(1, "Running"), pod(2, "Pending")), Seq.empty))
     k8s.getAllPodPhases.keySet shouldBe
       Set(KubernetesClient.generatePodName(1), KubernetesClient.generatePodName(2))
   }