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-7099-0f846092c61c2e9b3a0739be7507dc643a97607d in repository https://gitbox.apache.org/repos/asf/texera.git
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 + } +}
