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-6904-9de13865a0bda4f93041662f3ff736cdb8724900 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 49f9e2c414b116bf328bdb9154cc97dc4d1e7e9e Author: Meng Wang <[email protected]> AuthorDate: Sun Jul 26 20:47:43 2026 -0700 test(workflow-execution-service): add CollaborationResource unit tests (#6904) ### What changes were proposed in this PR? `CollaborationResource` had no spec and sat at 0% — all 87 tracked lines unhit. It reads as websocket-bound but the only collaborator is the `javax.websocket.Session` interface, which mocks cleanly with the ScalaMock already on amber's test classpath, so the session bookkeeping and message fan-out are ordinary unit-testable logic. Adds `CollaborationResourceSpec` with 13 tests covering the session lifecycle (`myOnOpen`, `myOnClose`), `WIdRequest` bookkeeping on both the authenticated and anonymous paths, `CommandRequest` and `RestoreVersionRequest` fan-out, `HeartBeatRequest`, and the locking branches that do not touch the database. A `mockSession` helper returns a `Session` with a fixed id whose outgoing messages are collected into a buffer; requests are built by serializing the real request case classes, so the `"type"` discriminator cannot drift out of sync with the production `@JsonTypeInfo` config. Three behaviors are worth calling out because they are subtle rather than obvious: - The multi-session `WIdRequest` test pins the `set.union(Set(senderSessId))` line, which returns a fresh set instead of mutating and only works because the result is reassigned into the map. - The fan-out tests assert the sender receives nothing, the same-workflow peer receives exactly one message, and a session on another workflow is untouched. - `wIdLockHolderSessionIdMap` stores a `null` sentinel for "no holder", which is a distinct state from an absent key. With the sentinel in place, `AcquireLockRequest` cannot resolve the holder session and rethrows — the spec pins that current behavior. The five bookkeeping maps live on the companion object and are therefore JVM-wide mutable state, so `beforeEach` clears all five; that is what keeps the suite order-independent. There are no clocks, threads, temp files or network calls in the spec. ### Any related issues, documentation, discussions? Closes #6900. The read-only `TryLockRequest` rejection and the lock hand-off inside `myOnClose` both reach `WorkflowAccessResource.hasWriteAccess` and therefore `SqlServer`; they are left uncovered rather than pulling `MockTexeraDB` (and an embedded Postgres process) in for two branches. ### How was this PR tested? `sbt "WorkflowExecutionService/testOnly *CollaborationResourceSpec"` passes with 13 tests; `Test/scalafmtCheck` and `Test/scalafix --check` are clean. For the failure path, the self-exclusion guard in the `CommandRequest` fan-out was temporarily removed from the production code, which reddened the fan-out test and exited non-zero, and the guard was then restored — so the assertions genuinely constrain the behavior. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-5) --------- Signed-off-by: Meng Wang <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../web/resource/CollaborationResourceSpec.scala | 297 +++++++++++++++++++++ 1 file changed, 297 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala new file mode 100644 index 0000000000..3c604578ad --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala @@ -0,0 +1,297 @@ +/* + * 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.resource + +import org.apache.texera.amber.util.JSONUtils +import org.apache.texera.dao.jooq.generated.tables.pojos.User +import org.apache.texera.web.model.collab.request._ +import org.apache.texera.web.resource.CollaborationResource._ +import org.scalamock.scalatest.MockFactory +import org.scalatest.BeforeAndAfterEach +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.util.concurrent.{Future => JFuture} +import javax.websocket.{RemoteEndpoint, Session} +import scala.collection.mutable.ArrayBuffer + +// Unit tests for CollaborationResource's session bookkeeping and message +// fan-out. The only collaborator is the javax.websocket.Session interface, so +// everything here runs in-process against mocks — no database, filesystem or +// network. The two branches that reach WorkflowAccessResource.hasWriteAccess +// (the read-only TryLockRequest rejection and the lock hand-off inside +// myOnClose) need SqlServer and are deliberately left uncovered. +class CollaborationResourceSpec + extends AnyFlatSpec + with Matchers + with MockFactory + with BeforeAndAfterEach { + + private var resource: CollaborationResource = _ + + // The five maps live on the companion object, i.e. they are JVM-wide mutable + // state shared by every test in the suite. Clearing them here is what keeps + // this spec order-independent. + override def beforeEach(): Unit = { + sessionIdSessionMap.clear() + sessionIdWIdMap.clear() + sessionIdUIdMap.clear() + wIdSessionIdsMap.clear() + wIdLockHolderSessionIdMap.clear() + resource = new CollaborationResource() + } + + /** + * A mocked Session whose getId is fixed and whose outgoing messages are + * collected into the returned buffer. `uId` seeds the authenticated user in + * the session's user properties; None models an anonymous session. + */ + private def mockSession(id: String, uId: Option[Int] = None): (Session, ArrayBuffer[String]) = { + val sent = ArrayBuffer[String]() + + val async = mock[RemoteEndpoint.Async] + (async + .sendText(_: String)) + .expects(*) + .onCall { (text: String) => + sent += text + null.asInstanceOf[JFuture[Void]] + } + .anyNumberOfTimes() + + val properties = new java.util.HashMap[String, Object]() + uId.foreach { uid => + val user = new User() + user.setUid(Integer.valueOf(uid)) + properties.put(classOf[User].getName, user) + } + + val session = mock[Session] + (() => session.getId).expects().returning(id).anyNumberOfTimes() + (() => session.getAsyncRemote).expects().returning(async).anyNumberOfTimes() + (() => session.getUserProperties).expects().returning(properties).anyNumberOfTimes() + + (session, sent) + } + + private def send(request: CollabWebSocketRequest): String = + JSONUtils.objectMapper.writeValueAsString(request) + + // -- session lifecycle ------------------------------------------------------ + + "myOnOpen" should "register the session" in { + val (session, _) = mockSession("s1") + + resource.myOnOpen(session) + + sessionIdSessionMap should contain key "s1" + sessionIdSessionMap("s1") shouldBe session + } + + "myOnClose" should "drop the session and its workflow bookkeeping" in { + val (session, _) = mockSession("s1") + resource.myOnOpen(session) + resource.myOnMsg(session, send(WIdRequest(7))) + sessionIdWIdMap should contain key "s1" + + resource.myOnClose(session) + + sessionIdSessionMap should not contain key("s1") + sessionIdWIdMap should not contain key("s1") + wIdSessionIdsMap(DUMMY_WID) shouldBe empty + } + + it should "leave the maps alone for a session that never sent a WIdRequest" in { + val (session, _) = mockSession("s1") + resource.myOnOpen(session) + + resource.myOnClose(session) + + sessionIdSessionMap shouldBe empty + sessionIdWIdMap shouldBe empty + wIdSessionIdsMap shouldBe empty + } + + // -- WIdRequest ------------------------------------------------------------- + + "WIdRequest" should "record the uid and the requested wid for an authenticated session" in { + val (session, _) = mockSession("s1", uId = Some(42)) + resource.myOnOpen(session) + + resource.myOnMsg(session, send(WIdRequest(7))) + + sessionIdUIdMap("s1") shouldBe 42 + sessionIdWIdMap("s1") shouldBe 7 + wIdSessionIdsMap(7) should contain only "s1" + } + + it should "fall back to DUMMY_WID for an anonymous session" in { + val (session, _) = mockSession("s1") + resource.myOnOpen(session) + + resource.myOnMsg(session, send(WIdRequest(7))) + + sessionIdWIdMap("s1") shouldBe DUMMY_WID + sessionIdUIdMap should not contain key("s1") + wIdSessionIdsMap(DUMMY_WID) should contain only "s1" + } + + it should "accumulate every session that joins the same wid" in { + val (first, _) = mockSession("s1", uId = Some(1)) + val (second, _) = mockSession("s2", uId = Some(2)) + resource.myOnOpen(first) + resource.myOnOpen(second) + + resource.myOnMsg(first, send(WIdRequest(7))) + resource.myOnMsg(second, send(WIdRequest(7))) + + // The union() call returns a fresh set rather than mutating in place, so + // this only holds because the result is reassigned into the map. + wIdSessionIdsMap(7) should contain theSameElementsAs Set("s1", "s2") + } + + // -- fan-out ---------------------------------------------------------------- + + /** + * Three open sessions: s1 and s2 share wid 1, s3 sits on wid 2. + */ + private def threeSessions(): ( + (Session, ArrayBuffer[String]), + (Session, ArrayBuffer[String]), + (Session, ArrayBuffer[String]) + ) = { + val first = mockSession("s1", uId = Some(1)) + val second = mockSession("s2", uId = Some(2)) + val third = mockSession("s3", uId = Some(3)) + List(first, second, third).foreach { case (session, _) => resource.myOnOpen(session) } + resource.myOnMsg(first._1, send(WIdRequest(1))) + resource.myOnMsg(second._1, send(WIdRequest(1))) + resource.myOnMsg(third._1, send(WIdRequest(2))) + (first, second, third) + } + + "CommandRequest" should "reach only the peers on the same workflow" in { + val ((sender, senderSent), (peer, peerSent), (other, otherSent)) = threeSessions() + senderSent.clear() + peerSent.clear() + otherSent.clear() + + resource.myOnMsg(sender, send(CommandRequest("do-something"))) + + peerSent should have size 1 + peerSent.head should include("CommandEvent") + peerSent.head should include("do-something") + senderSent shouldBe empty + otherSent shouldBe empty + peer.getId shouldBe "s2" + other.getId shouldBe "s3" + } + + "RestoreVersionRequest" should "reach only the peers on the same workflow" in { + val ((sender, senderSent), (_, peerSent), (_, otherSent)) = threeSessions() + senderSent.clear() + peerSent.clear() + otherSent.clear() + + resource.myOnMsg(sender, send(RestoreVersionRequest())) + + peerSent should have size 1 + peerSent.head should include("RestoreVersionEvent") + senderSent shouldBe empty + otherSent shouldBe empty + } + + // -- heartbeat -------------------------------------------------------------- + + "HeartBeatRequest" should "answer the sender only" in { + val ((sender, senderSent), (_, peerSent), _) = threeSessions() + senderSent.clear() + peerSent.clear() + + resource.myOnMsg(sender, send(HeartBeatRequest())) + + senderSent should have size 1 + senderSent.head should include("HeartBeatResponse") + peerSent shouldBe empty + } + + // -- locking ---------------------------------------------------------------- + + "TryLockRequest" should "grant the lock unconditionally on the DUMMY_WID workflow" in { + val (session, sent) = mockSession("s1") + resource.myOnOpen(session) + resource.myOnMsg(session, send(WIdRequest(7))) + sessionIdWIdMap("s1") shouldBe DUMMY_WID + sent.clear() + + resource.myOnMsg(session, send(TryLockRequest())) + + sent should have size 2 + sent.head should include("WorkflowAccessEvent") + sent.head should include("\"workflowReadonly\":false") + sent(1) should include("LockGrantedEvent") + } + + "AcquireLockRequest" should "hand the lock over from the previous holder" in { + val (holder, holderSent) = mockSession("s1", uId = Some(1)) + val (requester, requesterSent) = mockSession("s2", uId = Some(2)) + resource.myOnOpen(holder) + resource.myOnOpen(requester) + resource.myOnMsg(holder, send(WIdRequest(1))) + resource.myOnMsg(requester, send(WIdRequest(1))) + wIdLockHolderSessionIdMap(1) = "s1" + holderSent.clear() + requesterSent.clear() + + resource.myOnMsg(requester, send(AcquireLockRequest())) + + holderSent should have size 1 + holderSent.head should include("ReleaseLockEvent") + requesterSent should have size 1 + requesterSent.head should include("LockGrantedEvent") + wIdLockHolderSessionIdMap(1) shouldBe "s2" + } + + it should "re-grant the lock to the session that already holds it" in { + val (session, sent) = mockSession("s1", uId = Some(1)) + resource.myOnOpen(session) + resource.myOnMsg(session, send(WIdRequest(1))) + wIdLockHolderSessionIdMap(1) = "s1" + sent.clear() + + resource.myOnMsg(session, send(AcquireLockRequest())) + + sent should have size 1 + sent.head should include("LockGrantedEvent") + wIdLockHolderSessionIdMap(1) shouldBe "s1" + } + + it should "rethrow when the holder slot holds the null sentinel" in { + val (session, _) = mockSession("s1", uId = Some(1)) + resource.myOnOpen(session) + resource.myOnMsg(session, send(WIdRequest(1))) + // `null` means "no holder"; it is a distinct state from an absent key and + // the hand-off branch cannot look a null session id up. + wIdLockHolderSessionIdMap(1) = null + + a[NoSuchElementException] should be thrownBy + resource.myOnMsg(session, send(AcquireLockRequest())) + } +}
