(texera) branch main updated: test(amber): cover WorkflowExecution.getState PAUSED and UNKNOWN branches (#6140)

[gi...@apache.org]

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 92b7fc9168 test(amber): cover WorkflowExecution.getState PAUSED and UNKNOWN branches (#6140)
92b7fc9168 is described below

commit 92b7fc9168d79379a4871b9c9fd80fcb9df75041
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Jul 5 00:35:05 2026 -0700

    test(amber): cover WorkflowExecution.getState PAUSED and UNKNOWN branches (#6140)
    
    ### What changes were proposed in this PR?
    
    Extends `WorkflowExecutionSpec` to cover the two
    `WorkflowExecution.getState` branches the existing spec never exercised:
    
    - **PAUSED** — a running region whose operators are all paused;
    - **UNKNOWN** — a running operator with mixed worker states (one paused,
    one ready) that aggregates to `UNKNOWN`.
    
    Both are driven with the spec's existing region/operator/worker
    construction (a region carrying an incomplete port keeps its
    `RegionExecution` in `RUNNING`, so operator states are actually
    inspected). The `READY` return branch is intentionally not tested — it
    is dead code, since all-ready workers aggregate to `RUNNING`, so
    `OperatorExecution.getState` never returns `READY`.
    
    No production code is changed; this is test-only.
    
    ### Any related issues, documentation, discussions?
    
    Coverage gaps identified from the Codecov report for `apache/texera`.
    
    ### How was this PR tested?
    
    Extended ScalaTest spec, run locally:
    
    ```
    sbt "WorkflowExecutionService/testOnly *WorkflowExecutionSpec"
    ```
    
    All 14 pass. `scalafmt` and `scalafixAll --check` are clean.
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 4.8 [1M context])
---
 .../execution/WorkflowExecutionSpec.scala          | 48 +++++++++++++++++++++-
 1 file changed, 47 insertions(+), 1 deletion(-)

diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala
index 2bb9391332..d61ac8799b 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala
@@ -21,14 +21,16 @@ package org.apache.texera.amber.engine.architecture.coordinator.execution
 
 import org.apache.texera.amber.core.executor.OpExecInitInfo
 import org.apache.texera.amber.core.virtualidentity.{
+  ActorVirtualIdentity,
   ExecutionIdentity,
   OperatorIdentity,
   PhysicalOpIdentity,
   WorkflowIdentity
 }
-import org.apache.texera.amber.core.workflow.PhysicalOp
+import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PhysicalOp, PortIdentity}
 import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
 import org.apache.texera.amber.engine.architecture.scheduling.{Region, RegionIdentity}
+import org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState
 import org.scalatest.flatspec.AnyFlatSpec
 
 class WorkflowExecutionSpec extends AnyFlatSpec {
@@ -48,6 +50,26 @@ class WorkflowExecutionSpec extends AnyFlatSpec {
   private def region(regionId: Long, opId: String): Region =
     Region(RegionIdentity(regionId), Set(op(opId)), Set.empty)
 
+  /**
+    * A region carrying a single (input) port on `opId`. Because a freshly created
+    * `WorkerPortExecution` stays `completed = false`, the region's
+    * `RegionExecution.getState` remains RUNNING (never short-circuits to COMPLETED),
+    * so its operators' states are actually inspected by `WorkflowExecution.getState`.
+    */
+  private def regionWithPort(regionId: Long, opId: String): Region =
+    Region(
+      RegionIdentity(regionId),
+      Set(op(opId)),
+      Set.empty,
+      ports = Set(
+        GlobalPortIdentity(
+          opId = physicalOpId(opId),
+          portId = PortIdentity(id = 0, internal = false),
+          input = true
+        )
+      )
+    )
+
   "WorkflowExecution.initRegionExecution" should "create a new RegionExecution for the given region" in {
     val we = WorkflowExecution()
     val r = region(1, "a")
@@ -143,6 +165,30 @@ class WorkflowExecutionSpec extends AnyFlatSpec {
     assert(we.isCompleted)
   }
 
+  it should "return PAUSED when the only running region's operators are all paused" in {
+    val we = WorkflowExecution()
+    val regionExecution = we.initRegionExecution(regionWithPort(0, "a"))
+
+    val operatorExecution = regionExecution.initOperatorExecution(physicalOpId("a"))
+    val workerExecution = operatorExecution.initWorkerExecution(ActorVirtualIdentity("w0"))
+    workerExecution.update(1L, WorkerState.PAUSED)
+
+    assert(we.getState == WorkflowAggregatedState.PAUSED)
+    assert(!we.isCompleted)
+  }
+
+  it should "return UNKNOWN when a running operator has mixed worker states" in {
+    val we = WorkflowExecution()
+    val regionExecution = we.initRegionExecution(regionWithPort(0, "a"))
+
+    val operatorExecution = regionExecution.initOperatorExecution(physicalOpId("a"))
+    operatorExecution.initWorkerExecution(ActorVirtualIdentity("w0")).update(1L, WorkerState.PAUSED)
+    operatorExecution.initWorkerExecution(ActorVirtualIdentity("w1")).update(1L, WorkerState.READY)
+
+    assert(we.getState == WorkflowAggregatedState.UNKNOWN)
+    assert(!we.isCompleted)
+  }
+
   "WorkflowExecution.getLatestOperatorExecutionOption" should "return None when no operator execution exists for the id" in {
     val we = WorkflowExecution()
     we.initRegionExecution(region(0, "a"))