This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 8aec229032 test(computing-unit): cover the pod lookup, creation and
deletion paths (#7511)
8aec229032 is described below
commit 8aec22903261ddc69d43fe60def80e9b08547aa1
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Aug 10 10:30:02 2026 -0700
test(computing-unit): cover the pod lookup, creation and deletion paths
(#7511)
### What changes were proposed in this PR?
The spec covered the pure transforms and the namespace-wide wrappers and
stopped. The single-pod
half of the class — `getPodByName`, `podExists`, `getPodLimits`,
`createPod`, `deletePod` and the
pod URI — was untested, which is 42 of the file's 62 lines.
None of it needs a cluster: the fabric8 client is already a constructor
parameter, so the existing
Mockito fixture extends to the rest of the fluent chain, with the pod
that `createPod` builds
captured and inspected rather than sent anywhere.
Adds 8 tests. The three that matter:
- **the guard that refuses to overwrite a live pod** — creating over a
running unit would detach it
from its owner;
- **the `Option(...)` wrapper around the by-name lookup** — fabric8
returns `null` for an absent
pod rather than throwing, so the wrapper is all that stands between a
caller and an NPE;
- **the shared-memory volume appearing only when a size is requested** —
`/dev/shm` defaults to
64 Mi, too small for the Python workers, and the volume must not appear
when unrequested.
Also covered: the pod URI's service and namespace segments, the first
container's resource limits
and the empty-map fallback, env values reaching the container as
strings, and `deletePod` targeting
the cuid's own pod.
**Verified by mutation**, all reverted (production diff empty):
| Mutation | Result |
|---|---|
| pod name loses the cuid suffix | red |
| URI drops the namespace segment | red |
| `getPodByName` no longer null-guards | red |
| `createPod` overwrites an existing pod | red |
| cpu limit written from the memory value | red |
| pod hostname is not the pod name | red |
| `deletePod` targets a fixed cuid | red |
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7510
### How was this PR tested?
```
sbt "ComputingUnitManagingService/testOnly
org.apache.texera.service.util.KubernetesClientSpec"
```
```
[info] Total number of tests run: 16
[info] Tests: succeeded 16, failed 0, canceled 0, ignored 0, pending 0
```
8 new on top of the existing 8. `Test/scalafmtCheck` and `Test/scalafix
--check` both pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Co-authored-by: Meng Wang <[email protected]>
---
.../texera/service/util/KubernetesClientSpec.scala | 163 ++++++++++++++++++++-
1 file changed, 160 insertions(+), 3 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 f0d96347a9..f8e8353f7e 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
@@ -26,17 +26,29 @@ import io.fabric8.kubernetes.api.model.metrics.v1beta1.{
PodMetricsList,
PodMetricsListBuilder
}
-import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList,
PodListBuilder, Quantity}
+import io.fabric8.kubernetes.api.model.{
+ ContainerBuilder,
+ Pod,
+ PodBuilder,
+ PodList,
+ PodListBuilder,
+ Quantity,
+ ResourceRequirementsBuilder
+}
import io.fabric8.kubernetes.client.dsl.{
MetricAPIGroupDSL,
MixedOperation,
+ NamespaceableResource,
NonNamespaceOperation,
PodMetricOperation,
- PodResource
+ PodResource,
+ Resource
}
import io.fabric8.kubernetes.client.{KubernetesClient => Fabric8Client}
import org.apache.texera.common.config.KubernetesConfig
-import org.mockito.Mockito.{mock, when}
+import org.mockito.ArgumentCaptor
+import org.mockito.ArgumentMatchers.any
+import org.mockito.Mockito.{mock, times, verify, when}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
@@ -160,4 +172,149 @@ class KubernetesClientSpec extends AnyFlatSpec with
Matchers {
k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
k8s.getPodMetrics(999) shouldBe empty
}
+ // ── single-pod lookups, creation and deletion ──
+ // These reach the rest of the fluent chain: withName(...).get() for the
lookups,
+ // resource(pod).inNamespace(...).create() for creation, and .delete() for
removal. Everything
+ // is driven through the constructor seam, so no cluster is involved.
+
+ /** Extends the namespace stub with the by-name pod operations
`withName(...)` returns. */
+ private def clientWithNamedPod(podName: String, found: Pod): (Fabric8Client,
PodResource) = {
+ val client = stubbedClient(Seq.empty, Seq.empty)
+ val podsInNamespace = client.pods().inNamespace(namespace)
+ val podResource = mock(classOf[PodResource])
+ when(podsInNamespace.withName(podName)).thenReturn(podResource)
+ when(podResource.get()).thenReturn(found)
+ (client, podResource)
+ }
+
+ /** A pod carrying one container whose resource limits are set. */
+ private def podWithLimits(cuid: Int, limits: Map[String, String]): Pod =
+ new PodBuilder()
+ .withNewMetadata()
+ .withName(KubernetesClient.generatePodName(cuid))
+ .endMetadata()
+ .withNewSpec()
+ .addToContainers(
+ new ContainerBuilder()
+ .withName("main")
+ .withResources(
+ new ResourceRequirementsBuilder()
+ .withLimits(limits.map { case (k, v) => k -> new Quantity(v)
}.asJava)
+ .build()
+ )
+ .build()
+ )
+ .endSpec()
+ .build()
+
+ "generatePodURI" should "address the pod through its headless service inside
the namespace" in {
+ // The URI is how a computing unit is reached once it is up, so every
segment matters: a pod
+ // name alone, or the wrong namespace, resolves to nothing.
+ val uri = KubernetesClient.generatePodURI(7)
+ uri should startWith(KubernetesClient.generatePodName(7) + ".")
+ uri should
include(s".${KubernetesConfig.computeUnitServiceName}.$namespace.svc.cluster.local:")
+ uri should endWith(s":${KubernetesConfig.computeUnitPortNumber}")
+ }
+
+ "getPodByName" should "wrap a found pod and report a missing one as None" in
{
+ // fabric8 returns null rather than throwing for an absent pod, so the
Option() wrapper is the
+ // only thing standing between a caller and an NPE.
+ val name = KubernetesClient.generatePodName(1)
+ val (found, _) = clientWithNamedPod(name, pod(1, "Running"))
+ val (absent, _) = clientWithNamedPod(name, null)
+
+ new KubernetesClient(found).getPodByName(name).map(_.getMetadata.getName)
shouldBe Some(name)
+ new KubernetesClient(absent).getPodByName(name) shouldBe None
+ }
+
+ "podExists" should "follow the by-name lookup in both directions" in {
+ val name = KubernetesClient.generatePodName(2)
+ new KubernetesClient(clientWithNamedPod(name, pod(2,
"Running"))._1).podExists(2) shouldBe true
+ new KubernetesClient(clientWithNamedPod(name, null)._1).podExists(2)
shouldBe false
+ }
+
+ "getPodLimits" should "read the first container's limits and fall back to an
empty map" in {
+ val name = KubernetesClient.generatePodName(3)
+ val withLimits =
+ clientWithNamedPod(name, podWithLimits(3, Map("cpu" -> "2", "memory" ->
"4Gi")))._1
+ val missing = clientWithNamedPod(name, null)._1
+
+ new KubernetesClient(withLimits).getPodLimits(3) shouldBe Map("cpu" ->
"2", "memory" -> "4Gi")
+ new KubernetesClient(missing).getPodLimits(3) shouldBe empty
+ }
+
+ "createPod" should "refuse to overwrite a pod that already exists" in {
+ // Creating over a live unit would silently detach the running one from
its owner.
+ val name = KubernetesClient.generatePodName(4)
+ val k8s = new KubernetesClient(clientWithNamedPod(name, pod(4,
"Running"))._1)
+
+ val thrown = intercept[Exception] {
+ k8s.createPod(4, "1", "2Gi", "0", Map.empty)
+ }
+ thrown.getMessage should include("already exists")
+ }
+
+ it should "build the pod from the requested limits and env, and create it in
the namespace" in {
+ val name = KubernetesClient.generatePodName(5)
+ val (client, _) = clientWithNamedPod(name, null)
+ val namespaceable = mock(classOf[NamespaceableResource[Pod]])
+ val resource = mock(classOf[Resource[Pod]])
+ val captor = ArgumentCaptor.forClass(classOf[Pod])
+ when(client.resource(any(classOf[Pod]))).thenReturn(namespaceable)
+ when(namespaceable.inNamespace(namespace)).thenReturn(resource)
+ // create()'s return value is not asserted; the pod is inspected through
the captor below.
+ when(resource.create()).thenReturn(null)
+
+ new KubernetesClient(client).createPod(5, "2", "4Gi", "1", Map("UID" -> 9,
"MODE" -> "batch"))
+
+ verify(client).resource(captor.capture())
+ val built = captor.getValue
+ built.getSpec.getHostname shouldBe name
+ built.getSpec.getSubdomain shouldBe KubernetesConfig.computeUnitServiceName
+ val container = built.getSpec.getContainers.asScala.head
+ val limits = container.getResources.getLimits.asScala.map { case (k, v) =>
k -> v.toString }
+ limits("cpu") shouldBe "2"
+ limits("memory") shouldBe "4Gi"
+ // Env values arrive as Any and reach the container as strings.
+ container.getEnv.asScala.map(e => e.getName -> e.getValue).toMap shouldBe
+ Map("UID" -> "9", "MODE" -> "batch")
+ }
+
+ it should "mount a shared-memory volume only when a size is asked for" in {
+ // /dev/shm defaults to 64Mi in Kubernetes, which is too small for the
Python workers, so the
+ // volume is the fix — but it must not appear when no size was requested.
+ def build(shm: Option[String]): Pod = {
+ val name = KubernetesClient.generatePodName(6)
+ val (client, _) = clientWithNamedPod(name, null)
+ val namespaceable = mock(classOf[NamespaceableResource[Pod]])
+ val resource = mock(classOf[Resource[Pod]])
+ val captor = ArgumentCaptor.forClass(classOf[Pod])
+ when(client.resource(any(classOf[Pod]))).thenReturn(namespaceable)
+ when(namespaceable.inNamespace(namespace)).thenReturn(resource)
+ // create()'s return value is not asserted; the pod is inspected through
the captor below.
+ when(resource.create()).thenReturn(null)
+ new KubernetesClient(client).createPod(6, "1", "2Gi", "0", Map.empty,
shm)
+ verify(client).resource(captor.capture())
+ captor.getValue
+ }
+
+ val withShm = build(Some("1Gi"))
+ withShm.getSpec.getVolumes.asScala.map(_.getName) should contain("dshm")
+ withShm.getSpec.getVolumes.asScala
+ .find(_.getName == "dshm")
+ .flatMap(v => Option(v.getEmptyDir))
+ .map(_.getSizeLimit.toString) shouldBe Some("1Gi")
+
+
Option(build(None).getSpec.getVolumes).map(_.asScala.map(_.getName)).getOrElse(Nil)
should
+ not contain "dshm"
+ }
+
+ "deletePod" should "delete the pod for the cuid inside the namespace" in {
+ val name = KubernetesClient.generatePodName(8)
+ val (client, podResource) = clientWithNamedPod(name, pod(8, "Running"))
+
+ new KubernetesClient(client).deletePod(8)
+
+ verify(podResource, times(1)).delete()
+ }
}
