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 301e3a8769 test(amber): add websocket protocol serde specs (#7099)
301e3a8769 is described below
commit 301e3a8769438446c1f4e0e6a5dcd749d34be4da
Author: Xinyuan Lin <[email protected]>
AuthorDate: Wed Jul 29 20:59:07 2026 -0700
test(amber): add websocket protocol serde specs (#7099)
### What changes were proposed in this PR?
The websocket model tier had no test anywhere — `find amber/src/test
-path '*websocket*'` returned nothing. The uncovered classes are mostly
bare case classes, so the thing actually worth pinning is their Jackson
polymorphic registration, which is a **cross-language contract**:
| Where | What it says |
|---|---|
| `TexeraWebSocketRequest` / `TexeraWebSocketEvent` | `@JsonTypeInfo(use
= Id.NAME, include = As.PROPERTY, property = "type")` |
| every `@JsonSubTypes.Type` entry | no `name =` — and there is no
`@JsonTypeName` in the package |
| ⇒ resulting wire id | Jackson's `TypeNameIdResolver` fallback = **the
bare simple class name** |
| `workflow-websocket.interface.ts` | the Angular map keys on exactly
those strings |
So renaming a Scala class — or adding an explicit `name =` — silently
changes the wire protocol, with no compile error on either side. Nothing
guarded that.
```
client ──{"type":"ResultPaginationRequest", …}──▶
WorkflowWebsocketResource:81
objectMapper.readValue(msg, classOf[TexeraWebSocketRequest])
server ──{"type":"PaginatedResultEvent", …}──▶ SessionState:60
sendText(writeValueAsString(msg))
```
Three specs, all driving the **same `JSONUtils.objectMapper` production
uses** (a fresh `new ObjectMapper()` would test fiction — without
`DefaultScalaModule` none of these case classes bind at all):
- **`TexeraWebSocketRequestSpec`** — deserializes all 13 registered
subtypes through `classOf[TexeraWebSocketRequest]`; asserts the
registered id set *against the annotation itself*, so the expected list
is anchored to production rather than to another literal; asserts no
entry is explicitly named; rejects an unknown and a missing type id.
Also pins `ResultPaginationRequest`'s three default arguments — the live
path, since the client declares `columnOffset` / `columnLimit` /
`columnSearch` optional. If `DefaultScalaModule` were ever dropped,
`columnLimit` would bind to `0` and result pagination would silently
return zero columns.
- **`TexeraWebSocketEventSpec`** — events are **serialize-only** (server
→ client; nothing in `main` ever reads one back), so it asserts the
emitted discriminator for the 11 registered subtypes *and* the 5 that
are produced but unregistered, round-trips the 8 that are symmetric, and
pins that the effective inclusion rule is `NON_ABSENT` and not
`NON_EMPTY` (the result panel reads `updates` / `tableStats`
unconditionally).
- **`PaginatedResultEventSpec`** — the companion `apply`'s field
projection, with `pageIndex` ≠ `pageSize` and `requestID` ≠ `operatorID`
so any transposition fails.
Deliberately **not** asserted, and noted in-file so the reasoning
survives:
- `EditingTimeCompilationRequest.toLogicalPlanPojo` — no production
caller; the type stays in the registry assertions (registry membership
*is* the contract) but its dead method is not pinned.
- that the 5 unregistered events *fail* to deserialize — they do today,
but asserting a failure would turn this suite red the moment someone
harmlessly registers them.
Two divergences found while writing this, reported rather than encoded
as expectations:
- **`SkipTupleRequest` field-name mismatch** — the client sends `{
workers }` (`execute-workflow.service.ts:313`) while the case class
declares `workerIds`, and the shared mapper never disables
`FAIL_ON_UNKNOWN_PROPERTIES`, so the real client frame would be
rejected. Latent only because the handler throws `"skipping tuple is
temporarily disabled"` before reading the field.
- **`ExecutionResultService:498`** does `slice(columnOffset,
columnOffset + columnLimit)`; with the default `columnLimit =
Int.MaxValue` and any `columnOffset > 0` that sum overflows negative and
yields no columns. Unreachable today because the frontend's `useCache`
short-circuit keeps the all-columns request off the wire.
### Any related issues, documentation, discussions?
Closes #7097
### How was this PR tested?
Three new specs, 24 tests, run locally against Java 17:
```
sbt "WorkflowExecutionService/testOnly
org.apache.texera.web.model.websocket.request.TexeraWebSocketRequestSpec
org.apache.texera.web.model.websocket.event.TexeraWebSocketEventSpec
org.apache.texera.web.model.websocket.event.PaginatedResultEventSpec"
```
```
[info] Suites: completed 3, aborted 0
[info] Tests: succeeded 24, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
`Test/scalafmtCheck` and `Test/scalafix --check` both `[success]`. No
production file is touched — the diff is three new test files.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---
.../websocket/event/PaginatedResultEventSpec.scala | 103 ++++++
.../websocket/event/TexeraWebSocketEventSpec.scala | 352 +++++++++++++++++++++
.../request/TexeraWebSocketRequestSpec.scala | 304 ++++++++++++++++++
3 files changed, 759 insertions(+)
diff --git
a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
new file mode 100644
index 0000000000..aa48cc81c8
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
@@ -0,0 +1,103 @@
+/*
+ * 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.web.model.websocket.event
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.web.model.websocket.request.ResultPaginationRequest
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.jdk.CollectionConverters.IteratorHasAsScala
+
+/**
+ * Pins the request -> event projection that answers a pagination frame.
+ *
+ * `ExecutionResultService.handleResultPagination` builds its reply through
+ * `PaginatedResultEvent.apply(request, mappedResults, attributes)` on both
branches
+ * (the storage-backed one and the empty-storage fallback), so this companion
`apply`
+ * is the only place where a `ResultPaginationRequest` is turned into the
frame the
+ * result panel receives.
+ *
+ * Two things can silently break here and neither is a compile error:
+ * 1. `requestID` and `operatorID` are both `String` and adjacent in both
the request
+ * and the event, so swapping them type-checks. The frontend correlates
the reply
+ * to the pending request by `requestID`; a swap would make every page
load hang.
+ * The fixtures below use distinct values so a transposition fails.
+ * 2. The request carries four Ints (`pageIndex`, `pageSize`,
`columnOffset`,
+ * `columnLimit`) and only `pageIndex` is projected. Passing `pageSize`
instead
+ * type-checks too, so the fixture keeps `pageIndex` (3) far from
`pageSize` (25).
+ *
+ * The emitted field set is pinned as well: the Angular
`PaginatedResultEvent` declares
+ * exactly `requestID`, `operatorID`, `pageIndex`, `table` and `schema`, so
none of the
+ * request's column/pagination bookkeeping may leak onto the wire.
+ */
+class PaginatedResultEventSpec extends AnyFlatSpec with Matchers {
+
+ // pageIndex(3) != pageSize(25), requestID != operatorID, and the column
fields carry
+ // values that must NOT show up anywhere in the projected event.
+ private val request = ResultPaginationRequest(
+ requestID = "req-alpha",
+ operatorID = "op-beta",
+ pageIndex = 3,
+ pageSize = 25,
+ columnOffset = 4,
+ columnLimit = 6,
+ columnSearch = Some("city")
+ )
+
+ private val table = List(
+ objectMapper.createObjectNode().put("city", "Irvine"),
+ objectMapper.createObjectNode().put("city", "Anaheim")
+ )
+
+ private val schema = List(
+ new Attribute("city", AttributeType.STRING),
+ new Attribute("population", AttributeType.INTEGER)
+ )
+
+ "PaginatedResultEvent.apply" should "project requestID, operatorID and
pageIndex" in {
+ val event = PaginatedResultEvent(request, table, schema)
+ event.requestID shouldBe "req-alpha"
+ event.operatorID shouldBe "op-beta"
+ event.pageIndex shouldBe 3
+ }
+
+ it should "pass the table and schema through untouched" in {
+ val event = PaginatedResultEvent(request, table, schema)
+ event.table should contain theSameElementsInOrderAs table
+ event.schema should contain theSameElementsInOrderAs schema
+ }
+
+ "the projected event" should "serialize to exactly the fields the Angular
client declares" in {
+ val json = objectMapper.readTree(
+ objectMapper.writeValueAsString(PaginatedResultEvent(request, table,
schema))
+ )
+ json.fieldNames().asScala.toSet shouldBe
+ Set("type", "requestID", "operatorID", "pageIndex", "table", "schema")
+ json.get("type").asText() shouldBe "PaginatedResultEvent"
+ json.get("requestID").asText() shouldBe "req-alpha"
+ json.get("operatorID").asText() shouldBe "op-beta"
+ json.get("pageIndex").asInt() shouldBe 3
+ json.get("table").get(0).get("city").asText() shouldBe "Irvine"
+ json.get("schema").get(1).get("attributeName").asText() shouldBe
"population"
+ json.get("schema").get(1).get("attributeType").asText() shouldBe "integer"
+ }
+}
diff --git
a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
new file mode 100644
index 0000000000..6a1e063d59
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
@@ -0,0 +1,352 @@
+/*
+ * 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.web.model.websocket.event
+
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.google.protobuf.timestamp.Timestamp
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType}
+import org.apache.texera.amber.core.workflowruntimestate.{FatalErrorType,
WorkflowFatalError}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ ConsoleMessage,
+ ConsoleMessageType
+}
+import
org.apache.texera.amber.engine.architecture.rpc.controlreturns.{EvaluatedValue,
TypedValue}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.web.model.websocket.event.python.ConsoleUpdateEvent
+import
org.apache.texera.web.model.websocket.response.python.PythonExpressionEvaluateResponse
+import org.apache.texera.web.model.websocket.response.{
+ ClusterStatusUpdateEvent,
+ HeartBeatResponse,
+ ModifyLogicCompletedEvent,
+ ModifyLogicResponse,
+ RegionUpdateEvent
+}
+import org.apache.texera.web.service.ExecutionResultService.{
+ PaginationMode,
+ SetDeltaMode,
+ WebDataUpdate,
+ WebPaginationUpdate
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Instant
+
+/**
+ * Pins the server -> client half of the websocket wire contract.
+ *
+ * `SessionState.send` ships every event with
+ * `session.getAsyncRemote.sendText(objectMapper.writeValueAsString(msg))`,
where
+ * `msg: TexeraWebSocketEvent` and the mapper is `JSONUtils.objectMapper`.
This spec
+ * uses that exact mapper and that exact call so the emitted bytes are the
real ones.
+ *
+ * `TexeraWebSocketEvent` carries
+ * `@JsonTypeInfo(use = Id.NAME, include = As.PROPERTY, property = "type")`
and none of
+ * its `@JsonSubTypes.Type` entries supply `name =`, so the wire id is
Jackson's
+ * `TypeNameIdResolver` default: the BARE SIMPLE CLASS NAME. The Angular
client
+ * switches on those literals;
`frontend/src/app/workspace/types/workflow-websocket.interface.ts`
+ * spells the rule out:
+ *
+ * "Each type definition MUST follow the following rules:
+ * in either TexeraWebsocketRequestTypeMap or TexeraWebsocketEventTypeMap
+ * add a map entry:
+ * 1. key is the 'type' string, it must be the same as corresponding
backend class name
+ * 2. value is the payload this request/event needs"
+ *
+ * Its `TexeraWebsocketEventTypeMap` keys include `HeartBeatResponse`,
+ * `WorkflowStateEvent`, `OperatorStatisticsUpdateEvent`,
`WebResultUpdateEvent`,
+ * `WorkflowErrorEvent`, `ConsoleUpdateEvent`, `PaginatedResultEvent`,
+ * `CacheStatusUpdateEvent`, `PythonExpressionEvaluateResponse`,
+ * `WorkerAssignmentUpdateEvent`, `ModifyLogicResponse`,
`ModifyLogicCompletedEvent`,
+ * `ExecutionDurationUpdateEvent`, `ClusterStatusUpdateEvent`,
`RegionUpdateEvent` and
+ * `RegionStateEvent`. Renaming a Scala event class compiles fine on both
sides and
+ * silently drops the corresponding UI update -- that is what this spec
guards.
+ *
+ * Deliberate asymmetry: events are tested SERIALIZE-first. Five live event
classes
+ * (`ExecutionDurationUpdateEvent`, `RegionStateEvent`,
`ModifyLogicCompletedEvent`,
+ * `ClusterStatusUpdateEvent`, `RegionUpdateEvent`) are NOT listed in
`@JsonSubTypes`.
+ * They serialize correctly through the simple-name fallback, but
`typeFromId` cannot
+ * resolve them, so a blanket read-back loop would throw
`InvalidTypeIdException`.
+ * That is harmless today because those events are outbound only -- and it is
pinned
+ * below so nobody assumes the tier is symmetric.
+ */
+class TexeraWebSocketEventSpec extends AnyFlatSpec with Matchers {
+
+ private def write(e: TexeraWebSocketEvent): String =
objectMapper.writeValueAsString(e)
+
+ private def typeIdOf(e: TexeraWebSocketEvent): String =
+ objectMapper.readTree(write(e)).get("type").asText()
+
+ private val fatalError = WorkflowFatalError(
+ FatalErrorType.EXECUTION_FAILURE,
+ Timestamp(Instant.ofEpochSecond(1_700_000_000L)),
+ "message-1",
+ "details-2",
+ "op-3",
+ "worker-4"
+ )
+
+ private val consoleMessage = ConsoleMessage(
+ "worker-5",
+ Timestamp(Instant.ofEpochSecond(1_700_000_001L)),
+ ConsoleMessageType.PRINT,
+ "source-6",
+ "title-7",
+ "message-8"
+ )
+
+ private val metrics = OperatorAggregatedMetrics(
+ operatorState = "COMPLETED",
+ aggregatedInputRowCount = 11L,
+ aggregatedInputSize = 12L,
+ inputPortMetrics = Map("in-0" -> 13L),
+ aggregatedOutputRowCount = 14L,
+ aggregatedOutputSize = 15L,
+ outputPortMetrics = Map("out-0" -> 16L),
+ numWorkers = 17L,
+ aggregatedDataProcessingTime = 18L,
+ aggregatedControlProcessingTime = 19L,
+ aggregatedIdleTime = 20L
+ )
+
+ private val resultRow = objectMapper.createObjectNode().put("city", "Irvine")
+
+ /** The 11 events that ARE in `@JsonSubTypes`, one instance each. */
+ private val registeredEvents: List[(String, TexeraWebSocketEvent)] = List(
+ "HeartBeatResponse" -> HeartBeatResponse(),
+ "WorkflowErrorEvent" -> WorkflowErrorEvent(Seq(fatalError)),
+ "WorkflowStateEvent" -> WorkflowStateEvent("Running"),
+ "OperatorStatisticsUpdateEvent" ->
OperatorStatisticsUpdateEvent(Map("op-stats" -> metrics)),
+ "WebResultUpdateEvent" -> WebResultUpdateEvent(
+ Map("op-page" -> WebPaginationUpdate(PaginationMode(), 7L, List(1, 3))),
+ Map("op-page" -> Map("city" -> Map[String, Any]("distinct" -> 2)))
+ ),
+ "ConsoleUpdateEvent" -> ConsoleUpdateEvent("op-console",
Seq(consoleMessage)),
+ "CacheStatusUpdateEvent" -> CacheStatusUpdateEvent(Map("op-cache" ->
"cache valid")),
+ "PaginatedResultEvent" -> PaginatedResultEvent(
+ "req-1",
+ "op-2",
+ 3,
+ List(resultRow),
+ List(new Attribute("city", AttributeType.STRING))
+ ),
+ "PythonExpressionEvaluateResponse" -> PythonExpressionEvaluateResponse(
+ "len(tuple_)",
+ Seq(EvaluatedValue(Some(TypedValue("expr-1", "ref-2", "str-3", "type-4",
true)), Seq.empty))
+ ),
+ "WorkerAssignmentUpdateEvent" -> WorkerAssignmentUpdateEvent(
+ "op-assign",
+ Seq("worker-9", "worker-10")
+ ),
+ "ModifyLogicResponse" -> ModifyLogicResponse("op-modify", isValid = false,
"error-11")
+ )
+
+ /**
+ * Live events with a real producer that are NOT in `@JsonSubTypes`. The
producers are
+ * `ExecutionStatsService` (duration), `RegionExecutionManager` (region
state),
+ * `ExecutionReconfigurationService` (modify-logic completed),
`ClusterListener` /
+ * `WorkflowWebsocketResource` (cluster status) and `Coordinator` (region
update).
+ *
+ * `WorkflowAvailableResultEvent` is the sixth unregistered subtype but is
deliberately
+ * absent from this list: nothing in main constructs it, so pinning its
wire shape would
+ * only cement dead code.
+ */
+ private val outboundOnlyEvents: List[(String, TexeraWebSocketEvent)] = List(
+ "ExecutionDurationUpdateEvent" -> ExecutionDurationUpdateEvent(1234L,
isRunning = true),
+ "RegionStateEvent" -> RegionStateEvent(21L, "RUNNING"),
+ "ModifyLogicCompletedEvent" -> ModifyLogicCompletedEvent(List("op-22")),
+ "ClusterStatusUpdateEvent" -> ClusterStatusUpdateEvent(23),
+ "RegionUpdateEvent" -> RegionUpdateEvent(List((24L, List("op-25"))))
+ )
+
+ // The backend's registered set, spelled out rather than derived so a class
rename
+ // surfaces as a set diff. Deliberately not phrased as "what the client
resolves":
+ // the frontend's `TexeraWebsocketEventTypeMap` is a superset — it also keys
on the
+ // outbound-only events below (which reach it fine via the simple-name
fallback)
+ // plus a couple with no backend producer at all, e.g.
OperatorCurrentTuplesUpdateEvent
+ // and RecoveryStartedEvent.
+ private val expectedTypeIds: Set[String] = Set(
+ "HeartBeatResponse",
+ "WorkflowErrorEvent",
+ "WorkflowStateEvent",
+ "OperatorStatisticsUpdateEvent",
+ "WebResultUpdateEvent",
+ "ConsoleUpdateEvent",
+ "CacheStatusUpdateEvent",
+ "PaginatedResultEvent",
+ "PythonExpressionEvaluateResponse",
+ "WorkerAssignmentUpdateEvent",
+ "ModifyLogicResponse"
+ )
+
+ "TexeraWebSocketEvent @JsonSubTypes" should
+ "register exactly the wire type ids the backend declares" in {
+ val subTypes =
classOf[TexeraWebSocketEvent].getAnnotation(classOf[JsonSubTypes])
+ subTypes should not be null
+ subTypes.value().map(_.value().getSimpleName).toSet shouldBe
expectedTypeIds
+ }
+
+ it should "leave every subtype unnamed so the wire id stays the simple class
name" in {
+ val named = classOf[TexeraWebSocketEvent]
+ .getAnnotation(classOf[JsonSubTypes])
+ .value()
+ .filter(_.name().nonEmpty)
+ .map(t => s"${t.value().getSimpleName}=${t.name()}")
+ named.toList shouldBe empty
+ }
+
+ "every registered event" should "emit its simple class name as the type
property" in {
+ registeredEvents.map(_._1).toSet shouldBe expectedTypeIds
+ registeredEvents.foreach {
+ case (expectedId, event) =>
+ withClue(s"${event.getClass.getSimpleName}: ") {
+ typeIdOf(event) shouldBe expectedId
+ }
+ }
+ }
+
+ "every outbound-only event" should "still emit a type property via the
simple-name fallback" in {
+ // Not registered, yet the frontend keys on these ids -- serialization
must not
+ // silently drop "type" just because the subtype is missing from
@JsonSubTypes.
+ outboundOnlyEvents.foreach {
+ case (expectedId, event) =>
+ withClue(s"${event.getClass.getSimpleName}: ") {
+ typeIdOf(event) shouldBe expectedId
+ }
+ }
+ }
+
+ // Deliberately NOT asserted: that these five fail to deserialize through
the base
+ // trait. They do today (InvalidTypeIdException, since @JsonSubTypes has no
entry
+ // for them), but nothing in main ever reads an event back — the consumer is
+ // TypeScript — so pinning the failure would only make a future maintainer's
+ // harmless decision to register them turn this suite red.
+
+ "the fully symmetric events" should "survive a write/read round trip" in {
+ // Only the events whose payloads are plain Scala/Java values round-trip.
+ // WorkflowErrorEvent and ConsoleUpdateEvent cannot: their scalapb enums
+ // (FatalErrorType, ConsoleMessageType) have no Jackson creator.
WebResultUpdateEvent
+ // cannot either: WebResultUpdate is a sealed abstract class with no
@JsonTypeInfo.
+ val symmetricIds = Set(
+ "HeartBeatResponse",
+ "WorkflowStateEvent",
+ "OperatorStatisticsUpdateEvent",
+ "CacheStatusUpdateEvent",
+ "PaginatedResultEvent",
+ "PythonExpressionEvaluateResponse",
+ "WorkerAssignmentUpdateEvent",
+ "ModifyLogicResponse"
+ )
+ val symmetric = registeredEvents.filter { case (id, _) =>
symmetricIds.contains(id) }
+ symmetric.map(_._1).toSet shouldBe symmetricIds
+ symmetric.foreach {
+ case (id, event) =>
+ withClue(s"$id: ") {
+ objectMapper.readValue(write(event), classOf[TexeraWebSocketEvent])
shouldBe event
+ }
+ }
+ }
+
+ "WebResultUpdateEvent" should "emit empty maps rather than omitting the
keys" in {
+ // JSONUtils calls setSerializationInclusion twice (NON_NULL then
NON_ABSENT) and
+ // the second call replaces the first, so the effective rule is NON_ABSENT
— which
+ // subsumes NON_NULL but is NOT NON_EMPTY.
+ // The Angular WorkflowResultUpdateEvent reads `updates` and `tableStats`
+ // unconditionally, so tightening the inclusion rule would break the
result panel.
+ val json = objectMapper.readTree(write(WebResultUpdateEvent(Map.empty,
Map.empty)))
+ json.get("type").asText() shouldBe "WebResultUpdateEvent"
+ json.has("updates") shouldBe true
+ json.get("updates").isObject shouldBe true
+ json.get("updates").size() shouldBe 0
+ json.has("tableStats") shouldBe true
+ json.get("tableStats").size() shouldBe 0
+ }
+
+ it should "tag each nested WebResultUpdate mode with its @JsonTypeName" in {
+ // WebOutputMode has its own @JsonTypeInfo(Id.NAME) with @JsonTypeName on
each
+ // subtype, so the nested "type" is the annotated name, not the class name.
+ val paginated = objectMapper.readTree(
+ write(
+ WebResultUpdateEvent(
+ Map("op-page" -> WebPaginationUpdate(PaginationMode(), 7L, List(1,
3))),
+ Map.empty
+ )
+ )
+ )
+ val update = paginated.get("updates").get("op-page")
+ update.get("mode").get("type").asText() shouldBe "PaginationMode"
+ update.get("totalNumTuples").asLong() shouldBe 7L
+ update.get("dirtyPageIndices").toString shouldBe "[1,3]"
+
+ val delta = objectMapper.readTree(
+ write(
+ WebResultUpdateEvent(
+ Map("op-delta" -> WebDataUpdate(SetDeltaMode(), List.empty)),
+ Map.empty
+ )
+ )
+ )
+ delta.get("updates").get("op-delta").get("mode").get("type").asText()
shouldBe "SetDeltaMode"
+ }
+
+ "PythonExpressionEvaluateResponse" should "emit a value object even for an
empty proto Option" in {
+ // The only Option-typed field reachable from a registered event is
scalapb's
+ // EvaluatedValue.value. Include.NON_ABSENT would drop an Option.empty,
but scalapb's
+ // generated `getValue` accessor wins Jackson's bean introspection and
yields
+ // TypedValue.defaultInstance instead. The Angular EvaluatedValue declares
+ // `value: TypedValue` as REQUIRED and dereferences it, so the field must
stay
+ // present -- and must never be emitted as null.
+ val json = objectMapper.readTree(
+ write(PythonExpressionEvaluateResponse("1 + 1", Seq(EvaluatedValue(None,
Seq.empty))))
+ )
+ val value = json.get("values").get(0).get("value")
+ value.isNull shouldBe false
+ value.isObject shouldBe true
+ value.get("valueStr").asText() shouldBe ""
+ value.get("expandable").asBoolean() shouldBe false
+ }
+
+ "WorkflowErrorEvent" should "expose every fatal-error field the Angular
client reads" in {
+ // The TS WorkflowFatalError reads
message/details/operatorId/workerId/type.name and
+ // timestamp.{seconds,nanos}; all values below differ so a transposition
fails.
+ val error = objectMapper
+ .readTree(write(WorkflowErrorEvent(Seq(fatalError))))
+ .get("fatalErrors")
+ .get(0)
+ error.get("message").asText() shouldBe "message-1"
+ error.get("details").asText() shouldBe "details-2"
+ error.get("operatorId").asText() shouldBe "op-3"
+ error.get("workerId").asText() shouldBe "worker-4"
+ error.get("type").get("name").asText() shouldBe "EXECUTION_FAILURE"
+ error.get("timestamp").get("seconds").asLong() shouldBe 1_700_000_000L
+ }
+
+ "ConsoleUpdateEvent" should "expose every console-message field the Angular
client reads" in {
+ val message = objectMapper
+ .readTree(write(ConsoleUpdateEvent("op-console", Seq(consoleMessage))))
+ .get("messages")
+ .get(0)
+ message.get("workerId").asText() shouldBe "worker-5"
+ message.get("msgType").get("name").asText() shouldBe "PRINT"
+ message.get("source").asText() shouldBe "source-6"
+ message.get("title").asText() shouldBe "title-7"
+ message.get("message").asText() shouldBe "message-8"
+ }
+}
diff --git
a/amber/src/test/scala/org/apache/texera/web/model/websocket/request/TexeraWebSocketRequestSpec.scala
b/amber/src/test/scala/org/apache/texera/web/model/websocket/request/TexeraWebSocketRequestSpec.scala
new file mode 100644
index 0000000000..fc8199a0dd
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/model/websocket/request/TexeraWebSocketRequestSpec.scala
@@ -0,0 +1,304 @@
+/*
+ * 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.web.model.websocket.request
+
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.fasterxml.jackson.databind.exc.InvalidTypeIdException
+import org.apache.texera.amber.operator.limit.LimitOpDesc
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.web.model.websocket.request.python.{
+ DebugCommandRequest,
+ PythonExpressionEvaluateRequest
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Pins the client -> server half of the websocket wire contract.
+ *
+ * `WorkflowWebsocketResource.myOnMsg` deserializes every inbound frame with
+ * `objectMapper.readValue(message, classOf[TexeraWebSocketRequest])`, so
this spec
+ * uses the very same `JSONUtils.objectMapper` (DefaultScalaModule +
NoCtorDeserModule
+ * + Include.NON_ABSENT). A fresh `new ObjectMapper()` would test fiction:
without
+ * DefaultScalaModule none of the Scala case classes below bind at all.
+ *
+ * Why the discriminator strings are asserted literally:
`TexeraWebSocketRequest`
+ * carries `@JsonTypeInfo(use = Id.NAME, include = As.PROPERTY, property =
"type")`
+ * and every `@JsonSubTypes.Type` entry omits `name =`, so Jackson's
+ * `TypeNameIdResolver` falls back to the BARE SIMPLE CLASS NAME as the wire
id.
+ * The Angular client hard-codes those same strings; from
+ * `frontend/src/app/workspace/types/workflow-websocket.interface.ts`:
+ *
+ * "Each type definition MUST follow the following rules:
+ * in either TexeraWebsocketRequestTypeMap or TexeraWebsocketEventTypeMap
+ * add a map entry:
+ * 1. key is the 'type' string, it must be the same as corresponding
backend class name
+ * 2. value is the payload this request/event needs"
+ *
+ * The keys of `TexeraWebsocketRequestTypeMap` there are
`EditingTimeCompilationRequest`,
+ * `HeartBeatRequest`, `ModifyLogicRequest`, `ResultExportRequest`,
+ * `ResultPaginationRequest`, `RetryRequest`, `SkipTupleRequest`,
+ * `WorkflowExecuteRequest`, `WorkflowKillRequest`, `WorkflowPauseRequest`,
+ * `WorkflowCheckpointRequest`, `WorkflowResumeRequest`,
+ * `PythonExpressionEvaluateRequest` and `DebugCommandRequest`. Renaming a
Scala class
+ * in this package compiles cleanly on both sides and silently breaks the UI
at
+ * runtime -- that is the breakage this spec exists to catch.
+ *
+ * `ResultPaginationRequest`'s default arguments are the highest-value pin
here and sit
+ * on the live pagination path: the TS `PaginationRequest` declares
`columnOffset?`,
+ * `columnLimit?` and `columnSearch?` as OPTIONAL, so real frames omit them
and the
+ * server must fill in 0 / Int.MaxValue / None. Those defaults only
materialize because
+ * DefaultScalaModule calls the synthetic `$lessinit$greater$default$N`
methods -- drop
+ * that module and `columnLimit` binds to 0, so result pagination would
silently return
+ * zero columns for every request instead of failing loudly.
+ */
+class TexeraWebSocketRequestSpec extends AnyFlatSpec with Matchers {
+
+ private def read(json: String): TexeraWebSocketRequest =
+ objectMapper.readValue(json, classOf[TexeraWebSocketRequest])
+
+ private def frame(typeId: String, fields: String): String =
+ if (fields.isEmpty) s"""{"type":"$typeId"}""" else
s"""{"type":"$typeId",$fields}"""
+
+ // A LogicalPlanPojo / EditingTimeCompilationRequest payload with all four
lists empty.
+ private val emptyPlanFields =
+ """"operators":[],"links":[],"opsToViewResult":[],"opsToReuseResult":[]"""
+
+ // LogicalOp is itself polymorphic on a *different* property
("operatorType"), so a
+ // nested op exercises both discriminators in one frame.
+ private val limitOpJson = """{"operatorType":"Limit","limit":7}"""
+
+ private val executeFields =
+ s""""executionName":"exec-alpha","engineVersion":"engine-beta",""" +
+ s""""logicalPlan":{$emptyPlanFields},"workflowSettings":{},""" +
+ s""""emailNotificationEnabled":false,"computingUnitId":3"""
+
+ /**
+ * (wire type id, extra JSON fields, expected concrete class). The payloads
are
+ * built from each subtype's own declared field names, so the table pins
what the
+ * *server* accepts; the expected class proves the id resolved to the right
+ * subtype rather than to a same-shaped sibling.
+ *
+ * Note this is deliberately not a claim that every payload here matches
what the
+ * shipped Angular client sends. `SkipTupleRequest` is a known divergence —
the
+ * client sends `workers` (execute-workflow.service.ts) while the case class
+ * declares `workerIds`, and the shared mapper never disables
+ * FAIL_ON_UNKNOWN_PROPERTIES, so the real client frame would be rejected.
That
+ * is a production naming bug, not something to encode as an expectation
here;
+ * the feature is disabled anyway (ExecutionRuntimeService throws
+ * "skipping tuple is temporarily disabled" before reading the field).
+ */
+ private val registeredRequests: List[(String, String, Class[_ <:
TexeraWebSocketRequest])] =
+ List(
+ ("EditingTimeCompilationRequest", emptyPlanFields,
classOf[EditingTimeCompilationRequest]),
+ ("HeartBeatRequest", "", classOf[HeartBeatRequest]),
+ ("ModifyLogicRequest", s""""operator":$limitOpJson""",
classOf[ModifyLogicRequest]),
+ (
+ "ResultPaginationRequest",
+
""""requestID":"req-1","operatorID":"op-2","pageIndex":3,"pageSize":25""",
+ classOf[ResultPaginationRequest]
+ ),
+ ("RetryRequest", """"workers":["worker-1"]""", classOf[RetryRequest]),
+ ("SkipTupleRequest", """"workerIds":["worker-1"]""",
classOf[SkipTupleRequest]),
+ ("WorkflowExecuteRequest", executeFields,
classOf[WorkflowExecuteRequest]),
+ ("WorkflowKillRequest", "", classOf[WorkflowKillRequest]),
+ ("WorkflowPauseRequest", "", classOf[WorkflowPauseRequest]),
+ ("WorkflowResumeRequest", "", classOf[WorkflowResumeRequest]),
+ ("WorkflowCheckpointRequest", "", classOf[WorkflowCheckpointRequest]),
+ (
+ "PythonExpressionEvaluateRequest",
+ """"expression":"1 + 1","operatorId":"op-eval"""",
+ classOf[PythonExpressionEvaluateRequest]
+ ),
+ (
+ "DebugCommandRequest",
+ """"operatorId":"op-dbg","workerId":"worker-dbg","cmd":"break 12"""",
+ classOf[DebugCommandRequest]
+ )
+ )
+
+ // The 13 "type" strings the SERVER accepts, i.e. the backend registry.
Spelled out
+ // rather than derived so a rename shows up as a set diff instead of quietly
+ // re-deriving.
+ //
+ // Deliberately not phrased as "what the client sends": the frontend's
+ // `TexeraWebsocketRequestTypeMap` is a superset. It also declares
+ // `ResultExportRequest`, which on the backend is an HTTP model
+ // (web/model/http/request/result/, served by WorkflowExecutionsResource)
and not a
+ // TexeraWebSocketRequest subtype at all — see the divergence test below.
+ private val expectedTypeIds: Set[String] = Set(
+ "EditingTimeCompilationRequest",
+ "HeartBeatRequest",
+ "ModifyLogicRequest",
+ "ResultPaginationRequest",
+ "RetryRequest",
+ "SkipTupleRequest",
+ "WorkflowExecuteRequest",
+ "WorkflowKillRequest",
+ "WorkflowPauseRequest",
+ "WorkflowResumeRequest",
+ "WorkflowCheckpointRequest",
+ "PythonExpressionEvaluateRequest",
+ "DebugCommandRequest"
+ )
+
+ "TexeraWebSocketRequest @JsonSubTypes" should
+ "register exactly the wire type ids the server accepts" in {
+ val subTypes =
classOf[TexeraWebSocketRequest].getAnnotation(classOf[JsonSubTypes])
+ subTypes should not be null
+ subTypes.value().map(_.value().getSimpleName).toSet shouldBe
expectedTypeIds
+ }
+
+ it should "leave every subtype unnamed so the wire id stays the simple class
name" in {
+ // Adding `name = "..."` to any entry would change that subtype's wire id
without
+ // touching the class name, which the Angular map keys on.
+ val named = classOf[TexeraWebSocketRequest]
+ .getAnnotation(classOf[JsonSubTypes])
+ .value()
+ .filter(_.name().nonEmpty)
+ .map(t => s"${t.value().getSimpleName}=${t.name()}")
+ named.toList shouldBe empty
+ }
+
+ "every registered request type" should "deserialize through the polymorphic
base" in {
+ registeredRequests.map(_._1).toSet shouldBe expectedTypeIds
+ registeredRequests.foreach {
+ case (typeId, fields, expected) =>
+ withClue(s"""type id "$typeId": """) {
+ read(frame(typeId, fields)).getClass shouldBe expected
+ }
+ }
+ }
+
+ "an unknown type id" should "be rejected instead of silently ignored" in {
+ // A stale or typo'd client is a real path; it must fail loudly at the
mapper
+ // rather than bind to some same-shaped sibling. The id is deliberately
one that
+ // exists nowhere on either side, so the test cannot be misread as a claim
about
+ // any real type.
+ val ex =
intercept[InvalidTypeIdException](read("""{"type":"NoSuchWebSocketRequest"}"""))
+ ex.getMessage should include("NoSuchWebSocketRequest")
+ }
+
+ "the frontend-only ResultExportRequest id" should "not be accepted over the
websocket" in {
+ // The TS `TexeraWebsocketRequestTypeMap` declares `ResultExportRequest`,
but the
+ // backend class of that name is an HTTP model
(web/model/http/request/result/,
+ // reached through WorkflowExecutionsResource.exportResultToDataset) and
is not a
+ // TexeraWebSocketRequest subtype. Pinning the rejection documents the
divergence:
+ // if result export is ever moved onto the socket, this test is the
reminder that
+ // the subtype has to be registered too.
+ val ex =
intercept[InvalidTypeIdException](read("""{"type":"ResultExportRequest"}"""))
+ ex.getMessage should include("ResultExportRequest")
+ }
+
+ "a frame with no type property" should "be rejected" in {
+ val ex = intercept[InvalidTypeIdException](
+
read("""{"requestID":"req-1","operatorID":"op-2","pageIndex":3,"pageSize":25}""")
+ )
+ // The exception type is the contract here; the message check stays
deliberately
+ // loose (just the property name) because Jackson's exact wording is
version-specific.
+ ex.getMessage should include("type")
+ }
+
+ "ResultPaginationRequest" should "fill in the three optional column fields
when absent" in {
+ // The TS PaginationRequest marks these optional, so most real frames omit
them.
+ val req = read(
+ frame(
+ "ResultPaginationRequest",
+
""""requestID":"req-1","operatorID":"op-2","pageIndex":3,"pageSize":25"""
+ )
+ ).asInstanceOf[ResultPaginationRequest]
+ req.requestID shouldBe "req-1"
+ req.operatorID shouldBe "op-2"
+ req.pageIndex shouldBe 3
+ req.pageSize shouldBe 25
+ req.columnOffset shouldBe 0
+ req.columnLimit shouldBe Int.MaxValue
+ req.columnSearch shouldBe None
+ }
+
+ it should "let explicit column values override the defaults" in {
+ val req = read(
+ frame(
+ "ResultPaginationRequest",
+
""""requestID":"req-9","operatorID":"op-8","pageIndex":2,"pageSize":50,""" +
+ """"columnOffset":4,"columnLimit":6,"columnSearch":"city""""
+ )
+ ).asInstanceOf[ResultPaginationRequest]
+ req.columnOffset shouldBe 4
+ req.columnLimit shouldBe 6
+ req.columnSearch shouldBe Some("city")
+ }
+
+ "WorkflowExecuteRequest" should "bind an absent replayFromExecution to None"
in {
+ val req = read(frame("WorkflowExecuteRequest", executeFields))
+ .asInstanceOf[WorkflowExecuteRequest]
+ req.replayFromExecution shouldBe None
+ // distinct values so a transposed executionName/engineVersion binding
fails here
+ req.executionName shouldBe "exec-alpha"
+ req.engineVersion shouldBe "engine-beta"
+ req.computingUnitId shouldBe 3
+ req.emailNotificationEnabled shouldBe false
+ }
+
+ it should "bind a present replayFromExecution to Some with a nested eid and
interaction" in {
+ // The replay path is the reason WorkflowExecuteRequest has an Option at
all:
+ // an absent key must stay None (asserted above) while a present object
must
+ // populate both nested fields. The values differ in kind so a swapped
+ // eid/interaction binding cannot pass.
+ val req = read(
+ frame(
+ "WorkflowExecuteRequest",
+ executeFields +
""","replayFromExecution":{"eid":91,"interaction":"interaction-7"}"""
+ )
+ ).asInstanceOf[WorkflowExecuteRequest]
+ req.replayFromExecution shouldBe Some(ReplayExecutionInfo(91L,
"interaction-7"))
+ }
+
+ "the python request payloads" should "bind each wire field to the matching
parameter" in {
+ // All values differ, so any transposition (or a renamed JSON property)
fails.
+ val evaluate = read(
+ frame(
+ "PythonExpressionEvaluateRequest",
+ """"expression":"len(tuple_)","operatorId":"op-eval""""
+ )
+ ).asInstanceOf[PythonExpressionEvaluateRequest]
+ evaluate.expression shouldBe "len(tuple_)"
+ evaluate.operatorId shouldBe "op-eval"
+
+ val debug = read(
+ frame(
+ "DebugCommandRequest",
+ """"operatorId":"op-dbg","workerId":"worker-dbg","cmd":"break 12""""
+ )
+ ).asInstanceOf[DebugCommandRequest]
+ debug.operatorId shouldBe "op-dbg"
+ debug.workerId shouldBe "worker-dbg"
+ debug.cmd shouldBe "break 12"
+ }
+
+ "ModifyLogicRequest" should "resolve the nested LogicalOp discriminator too"
in {
+ // "type" selects the request, "operatorType" selects the operator; the two
+ // @JsonTypeInfo properties must not collide.
+ val req = read(frame("ModifyLogicRequest", s""""operator":$limitOpJson"""))
+ .asInstanceOf[ModifyLogicRequest]
+ req.operator shouldBe a[LimitOpDesc]
+ req.operator.asInstanceOf[LimitOpDesc].limit shouldBe 7
+ }
+}
