(texera) branch gh-readonly-queue/main/pr-7441-12eaccbb24884fccf1dba26d15fff7ceb1db4aba created (now 08a2eac8eb)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7441-12eaccbb24884fccf1dba26d15fff7ceb1db4aba
in repository https://gitbox.apache.org/repos/asf/texera.git


      at 08a2eac8eb test(amber): cover ExecutionConsoleService's console routing and debug commands (#7441)

This branch includes the following new commits:

     new 08a2eac8eb test(amber): cover ExecutionConsoleService's console routing and debug commands (#7441)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(texera) 01/01: test(amber): cover ExecutionConsoleService's console routing and debug commands (#7441)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7441-12eaccbb24884fccf1dba26d15fff7ceb1db4aba
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 08a2eac8eb7151eaac33b18cd8a47a20599e5fa7
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Aug 9 16:49:50 2026 -0700

    test(amber): cover ExecutionConsoleService's console routing and debug commands (#7441)
    
    ### What changes were proposed in this PR?
    
    The suite covered the `ConsoleMessageProcessor` object; the service
    class around it was untouched. That class owns the console diff the
    frontend is driven from, the worker-to-operator keying that decides
    where a message lands, and the websocket handler behind the debugger.
    
    Converts the spec to a TestKit suite and adds 8 tests:
    
    - a **debugger** message is never truncated, while an ordinary one still
    is — the debugger's output is the frame the user asked to see
    - the diff reports **only messages added since the last state**; the
    frontend appends what it is sent, so emitting the whole buffer would
    duplicate every earlier line on each update
    - a console message is filed under the **logical** operator id — the
    worker id carries the physical layer and worker index, and anything else
    strands the output where the frontend will not look for it
    - a debug command is attributed to `USER-<uid>`, falls back to
    `USER-UNKNOWN` with no session user, is filed under the operator rather
    than the worker, and is forwarded to the coordinator with the worker it
    names
    
    **Verified by mutation**, all reverted (production diff empty):
    
    | Mutation | Result |
    |---|---|
    | truncate debugger messages too | red |
    | send the whole console buffer instead of the diff | red |
    | file the message under the physical op id | red |
    | file the message under the raw worker id | red |
    | attribute every debug command to a constant | red |
    | file the debug command under the worker | red |
    | forward the wrong id to the coordinator | red |
    | drop the command from the message title | red |
    
    Everything runs on an empty-plan `AmberClient` with a mocked
    coordinator: no engine, database or Iceberg storage. The Iceberg-backed
    writer path (`getOrCreateWriter` and the execution-state commit loop) is
    `private` and storage-bound; it is left uncovered rather than padded
    with a no-throw test, and the spec says so.
    
    One note recorded in the spec: the console store's event observable
    replays on subscribe, so the diff test subscribes first and asserts on
    the batch published for the second message.
    
    No production file is touched.
    
    ### Any related issues, documentation, discussions?
    
    Closes #7438
    
    ### How was this PR tested?
    
    ```
    sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionConsoleServiceSpec"
    ```
    
    ```
    [info] Tests: succeeded 13, failed 0, canceled 0, ignored 0, pending 0
    [info] All tests passed.
    ```
    
    8 new on top of the existing 5. `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)
    
    ---------
    
    Signed-off-by: Xinyuan Lin <xi...@uci.edu>
    Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
 .../web/service/ExecutionConsoleServiceSpec.scala  | 218 ++++++++++++++++++++-
 1 file changed, 214 insertions(+), 4 deletions(-)

diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala
index d4753984cf..b1d647d035 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala
@@ -20,17 +20,57 @@
 package org.apache.texera.web.service
 
 import com.google.protobuf.timestamp.Timestamp
+import com.twitter.util.{Future => TwitterFuture}
+import io.reactivex.rxjava3.disposables.Disposable
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.testkit.TestKit
+import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext}
+import org.apache.texera.amber.engine.architecture.coordinator.CoordinatorConfig
 import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
   ConsoleMessage,
-  ConsoleMessageType
+  ConsoleMessageType,
+  DebugCommandRequest => AmberDebugCommandRequest
 }
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn
+import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceFs2Grpc
+import org.apache.texera.amber.engine.common.client.AmberClient
 import org.apache.texera.amber.engine.common.executionruntimestate.ExecutionConsoleStore
-import org.scalatest.flatspec.AnyFlatSpec
+import org.apache.texera.web.WebsocketInput
+import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent
+import org.apache.texera.web.model.websocket.event.python.ConsoleUpdateEvent
+import org.apache.texera.web.model.websocket.request.python.DebugCommandRequest
+import org.apache.texera.web.storage.ExecutionStateStore
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpecLike
 import org.scalatest.matchers.should.Matchers
 
 import java.time.Instant
-
-class ExecutionConsoleServiceSpec extends AnyFlatSpec with Matchers {
+import scala.collection.mutable.ListBuffer
+import scala.reflect.ClassTag
+
+/**
+  * The `ConsoleMessageProcessor` object is covered by the first half of this suite. The service
+  * class around it was entirely uncovered: it owns the console diff that decides what the frontend
+  * is told, the worker-to-operator keying that decides where a message lands, and the websocket
+  * handlers behind the debugger.
+  *
+  * Everything here runs on an empty-plan AmberClient with a mocked coordinator, so no engine is
+  * involved. Note that ExecutionConsoleService still schedules asynchronous console persistence
+  * (writer creation + operator-executions insertion); this suite focuses on console routing/diffing
+  * and does not assert on the persistence side effects.
+  */
+class ExecutionConsoleServiceSpec
+    extends TestKit(ActorSystem("ExecutionConsoleServiceSpec"))
+    with AnyFlatSpecLike
+    with Matchers
+    with MockFactory
+    with BeforeAndAfterAll {
+
+  override def afterAll(): Unit = {
+    try TestKit.shutdownActorSystem(system)
+    finally super.afterAll()
+  }
 
   // Constants for testing
   val standardBufferSize: Int = 100
@@ -231,4 +271,174 @@ class ExecutionConsoleServiceSpec extends AnyFlatSpec with Matchers {
     val expectedTruncatedTitle = "a" * (messageDisplayLength - 3) + "..."
     opInfo.consoleMessages.head.title shouldBe expectedTruncatedTitle
   }
+  // ---------------------------------------------------------------- instance
+
+  /** Empty-plan client that captures the ConsoleMessage callback the service registers. */
+  private final class TestAmberClient(
+      override val coordinatorInterface: CoordinatorServiceFs2Grpc[TwitterFuture, Unit]
+  ) extends AmberClient(
+        system,
+        new WorkflowContext(),
+        PhysicalPlan(Set.empty, Set.empty),
+        CoordinatorConfig(None, None, None, None),
+        _ => ()
+      ) {
+    var consoleCallback: ConsoleMessage => Unit = _
+
+    override def registerCallback[T](callback: T => Unit)(implicit ct: ClassTag[T]): Disposable = {
+      if (ct.runtimeClass == classOf[ConsoleMessage]) {
+        consoleCallback = callback.asInstanceOf[ConsoleMessage => Unit]
+      }
+      Disposable.empty()
+    }
+
+    def dispose(): Unit = super.shutdown()
+  }
+
+  private final class Fixture(
+      val client: TestAmberClient,
+      val coordinator: CoordinatorServiceFs2Grpc[TwitterFuture, Unit],
+      val stateStore: ExecutionStateStore,
+      val wsInput: WebsocketInput,
+      val service: ExecutionConsoleService
+  ) {
+    def close(): Unit = {
+      service.unsubscribeAll()
+      client.dispose()
+    }
+  }
+
+  private def fixture(): Fixture = {
+    val coordinator = mock[CoordinatorServiceFs2Grpc[TwitterFuture, Unit]]
+    val client = new TestAmberClient(coordinator)
+    val stateStore = new ExecutionStateStore
+    val wsInput = new WebsocketInput(ListBuffer.empty[Throwable] += _)
+    val service = new ExecutionConsoleService(client, stateStore, wsInput, new WorkflowContext())
+    new Fixture(client, coordinator, stateStore, wsInput, service)
+  }
+
+  private def message(
+      workerId: String = "Worker:WF1-udf1-main-0",
+      title: String = "hello",
+      msgType: ConsoleMessageType = ConsoleMessageType.PRINT
+  ): ConsoleMessage =
+    new ConsoleMessage(workerId, Timestamp(Instant.now), msgType, "src", title, "content")
+
+  private def withFixture(body: Fixture => Unit): Unit = {
+    val f = fixture()
+    try body(f)
+    finally f.close()
+  }
+
+  "processConsoleMessage" should "leave a debugger message untouched however long it is" in {
+    // The debugger's output is the payload the user asked to see; truncating it would cut off the
+    // frame or variable they are inspecting.
+    withFixture { f =>
+      val long = "a" * (f.service.consoleMessageDisplayLength + 50)
+
+      val processed = f.service.processConsoleMessage(
+        message(title = long, msgType = ConsoleMessageType.DEBUGGER)
+      )
+
+      processed.title shouldBe long
+    }
+  }
+
+  it should "still truncate an ordinary message to the configured length" in {
+    withFixture { f =>
+      val long = "a" * (f.service.consoleMessageDisplayLength + 50)
+
+      val processed = f.service.processConsoleMessage(message(title = long))
+
+      processed.title.length shouldBe f.service.consoleMessageDisplayLength
+      processed.title should endWith("...")
+    }
+  }
+
+  "the console diff handler" should "report only the messages added since the last state" in {
+    // The frontend appends what it is sent. Emitting the whole buffer instead of the delta would
+    // duplicate every earlier line on each update.
+    withFixture { f =>
+      // One batch is published per state update. Subscribing up front and reading the batch for the
+      // SECOND message is what shows the delta: the first message must not appear in it again.
+      val batches = ListBuffer.empty[Iterable[TexeraWebSocketEvent]]
+      val sub = f.stateStore.consoleStore.getWebsocketEventObservable
+        .subscribe((batch: Iterable[TexeraWebSocketEvent]) => batches += batch)
+
+      try {
+        f.client.consoleCallback(message(title = "first"))
+        f.client.consoleCallback(message(title = "second"))
+      } finally sub.dispose()
+
+      val titles =
+        batches.last.collect { case e: ConsoleUpdateEvent => e.messages.map(_.title) }.flatten
+      titles shouldBe Seq("second")
+    }
+  }
+
+  "the console message callback" should "file a message under the logical operator id" in {
+    // The worker id carries the physical layer and worker index; the frontend console is keyed by
+    // the logical operator, so anything else silently strands the output.
+    withFixture { f =>
+      f.client.consoleCallback(message(workerId = "Worker:WF1-udf1-main-0"))
+
+      f.stateStore.consoleStore.getState.operatorConsole.keys should contain("udf1")
+    }
+  }
+
+  it should "store the truncated form, not the original" in {
+    withFixture { f =>
+      val long = "b" * (f.service.consoleMessageDisplayLength + 50)
+
+      f.client.consoleCallback(message(title = long))
+
+      val stored = f.stateStore.consoleStore.getState.operatorConsole("udf1").consoleMessages
+      stored.map(_.title.length) shouldBe Seq(f.service.consoleMessageDisplayLength)
+    }
+  }
+
+  "a debug command" should "be attributed to the user that issued it" in {
+    withFixture { f =>
+      (f.coordinator.debugCommand _)
+        .expects(AmberDebugCommandRequest("Worker:WF1-udf1-main-0", "break 12"), ())
+        .returning(TwitterFuture.value(EmptyReturn()))
+
+      f.wsInput.onNext(
+        DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "break 12"),
+        Some(7)
+      )
+
+      val stored = f.stateStore.consoleStore.getState.operatorConsole("udf1").consoleMessages
+      stored.map(_.source) shouldBe Seq("USER-7")
+      stored.map(_.title) shouldBe Seq("break 12")
+    }
+  }
+
+  it should "fall back to UNKNOWN when there is no session user" in {
+    withFixture { f =>
+      (f.coordinator.debugCommand _)
+        .expects(*, *)
+        .returning(TwitterFuture.value(EmptyReturn()))
+
+      f.wsInput.onNext(DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "cont"), None)
+
+      f.stateStore.consoleStore.getState
+        .operatorConsole("udf1")
+        .consoleMessages
+        .map(_.source) shouldBe Seq("USER-UNKNOWN")
+    }
+  }
+
+  it should "file the command under the operator, not the worker" in {
+    // req carries both; keying by workerId would scatter the command across per-worker consoles.
+    withFixture { f =>
+      (f.coordinator.debugCommand _).expects(*, *).returning(TwitterFuture.value(EmptyReturn()))
+
+      f.wsInput.onNext(DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "cont"), Some(1))
+
+      val keys = f.stateStore.consoleStore.getState.operatorConsole.keys
+      keys should contain("udf1")
+      keys should not contain "Worker:WF1-udf1-main-0"
+    }
+  }
 }