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-7166-0022e353c9909d6001421a15ed3d7327ca486a0b in repository https://gitbox.apache.org/repos/asf/texera.git
commit d9c443f38e2f4ac730c840a9fbb0ed8ed19cde63 Author: Meng Wang <[email protected]> AuthorDate: Fri Jul 31 11:48:48 2026 -0700 test(amber): extend WorkflowVersionResource unit test coverage (#7166) ### What changes were proposed in this PR? Extends the existing `MockTexeraDB`-backed `WorkflowVersionResourceSpec` (which only covered `fetchSubsequentVersions` + `applyPatch`) to cover the resource's remaining methods. 14 new tests: - **Pure diff logic** — `applyPatch` reconstructs an older content from a stored delta (no DB); `isSnapshotInRangeUnimportant` returns true for equal bounds and for a replace-only range, false when the range contains an important (non-replace) change. - **Version persistence (object methods)** — `insertNewVersion` persists a retrievable version; `getLatestVersion` returns the highest vid, and creates one when the workflow has none; `insertVersion` always versions a new workflow, and otherwise only when the content actually changed. - **Resource endpoints** — `retrieveVersionsOfWorkflow` returns the importance-encoded list for a user with access and an empty list otherwise; `retrieveWorkflowVersion` reconstructs the workflow for an authorized user and throws `ForbiddenException` otherwise; `cloneVersion` creates a new workflow from an existing version. The spec's `beforeEach` now also seeds an owner + a stranger user with the appropriate access grants (mirroring `WorkflowResourceCoverSpec`) so the auth-gated endpoints can be exercised. No production code was changed. ### Any related issues, documentation, discussions? Closes #7161. ### How was this PR tested? `sbt "WorkflowExecutionService/testOnly *WorkflowVersionResourceSpec"` — 15 passed (1 existing + 14 new), run twice for determinism against the embedded Postgres. The failure path was verified by breaking an assertion (1 failed, non-zero exit), and `Test/scalafmtCheck` + `Test/scalafix --check` are clean. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context]) --- .../workflow/WorkflowVersionResourceSpec.scala | 282 ++++++++++++++++++++- 1 file changed, 272 insertions(+), 10 deletions(-) diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala index 3e86f38c2e..06d933f51d 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala @@ -20,30 +20,56 @@ package org.apache.texera.web.resource.dashboard.user.workflow import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.texera.auth.SessionUser import org.apache.texera.dao.MockTexeraDB import org.apache.texera.dao.jooq.generated.Tables -import org.apache.texera.dao.jooq.generated.tables.daos.{WorkflowDao, WorkflowVersionDao} -import org.apache.texera.dao.jooq.generated.tables.pojos.{Workflow, WorkflowVersion} +import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowDao, + WorkflowOfUserDao, + WorkflowUserAccessDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowOfUser, + WorkflowUserAccess, + WorkflowVersion +} import org.jooq.impl.DSL import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} import java.sql.Timestamp import java.util.UUID import java.util.concurrent.TimeUnit +import javax.ws.rs.ForbiddenException import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters._ class WorkflowVersionResourceSpec extends AnyFlatSpec + with Matchers with BeforeAndAfterAll with BeforeAndAfterEach with MockTexeraDB { private val testWorkflowWid = 2000 + scala.util.Random.nextInt(1000) + private val ownerUid = 4000 + scala.util.Random.nextInt(1000) + private val strangerUid = 6000 + scala.util.Random.nextInt(1000) private var testWorkflow: Workflow = _ + private var owner: User = _ + private var stranger: User = _ private var workflowDao: WorkflowDao = _ private var workflowVersionDao: WorkflowVersionDao = _ + private var userDao: UserDao = _ + private var workflowOfUserDao: WorkflowOfUserDao = _ + private var workflowUserAccessDao: WorkflowUserAccessDao = _ + private var resource: WorkflowVersionResource = _ private val capturedVersions = ArrayBuffer.empty[Integer] @@ -52,17 +78,36 @@ class WorkflowVersionResourceSpec } override protected def beforeEach(): Unit = { + workflowDao = new WorkflowDao(getDSLContext.configuration()) + workflowVersionDao = new WorkflowVersionDao(getDSLContext.configuration()) + userDao = new UserDao(getDSLContext.configuration()) + workflowOfUserDao = new WorkflowOfUserDao(getDSLContext.configuration()) + workflowUserAccessDao = new WorkflowUserAccessDao(getDSLContext.configuration()) + resource = new WorkflowVersionResource() + testWorkflow = new Workflow testWorkflow.setWid(Integer.valueOf(testWorkflowWid)) testWorkflow.setName("test_workflow_" + UUID.randomUUID().toString.substring(0, 8)) testWorkflow.setContent(createWorkflowContent("initial")) testWorkflow.setDescription("test description") + testWorkflow.setIsPublic(false) + testWorkflow.setCreationTime(new Timestamp(System.currentTimeMillis())) + testWorkflow.setLastModifiedTime(new Timestamp(System.currentTimeMillis())) - workflowDao = new WorkflowDao(getDSLContext.configuration()) - workflowVersionDao = new WorkflowVersionDao(getDSLContext.configuration()) + owner = makeUser(ownerUid, "wfv_owner") + stranger = makeUser(strangerUid, "wfv_stranger") cleanupTestData() + userDao.insert(owner) + userDao.insert(stranger) workflowDao.insert(testWorkflow) + + val ownership = new WorkflowOfUser + ownership.setUid(Integer.valueOf(ownerUid)) + ownership.setWid(Integer.valueOf(testWorkflowWid)) + workflowOfUserDao.insert(ownership) + grantAccess(ownerUid, PrivilegeEnum.WRITE) + capturedVersions.clear() } @@ -71,14 +116,36 @@ class WorkflowVersionResourceSpec } private def cleanupTestData(): Unit = { - getDSLContext + val ctx = getDSLContext + // Purge the fixed test workflow plus anything our test users came to own + // (cloneVersion creates a fresh workflow), children before parents so the + // user rows can be removed without violating foreign keys. + val wids = (ctx + .select(Tables.WORKFLOW_OF_USER.WID) + .from(Tables.WORKFLOW_OF_USER) + .where( + Tables.WORKFLOW_OF_USER.UID.in(Integer.valueOf(ownerUid), Integer.valueOf(strangerUid)) + ) + .fetchInto(classOf[Integer]) + .asScala + .toSet + Integer.valueOf(testWorkflowWid)).toList + + ctx .deleteFrom(Tables.WORKFLOW_VERSION) - .where(Tables.WORKFLOW_VERSION.WID.eq(testWorkflowWid)) + .where(Tables.WORKFLOW_VERSION.WID.in(wids: _*)) .execute() - - getDSLContext - .deleteFrom(Tables.WORKFLOW) - .where(Tables.WORKFLOW.WID.eq(testWorkflowWid)) + ctx + .deleteFrom(Tables.WORKFLOW_USER_ACCESS) + .where(Tables.WORKFLOW_USER_ACCESS.WID.in(wids: _*)) + .execute() + ctx + .deleteFrom(Tables.WORKFLOW_OF_USER) + .where(Tables.WORKFLOW_OF_USER.WID.in(wids: _*)) + .execute() + ctx.deleteFrom(Tables.WORKFLOW).where(Tables.WORKFLOW.WID.in(wids: _*)).execute() + ctx + .deleteFrom(Tables.USER) + .where(Tables.USER.UID.in(Integer.valueOf(ownerUid), Integer.valueOf(strangerUid))) .execute() } @@ -86,6 +153,25 @@ class WorkflowVersionResourceSpec closeConnectionPool() } + private def makeUser(uid: Int, name: String): User = { + val user = new User + user.setUid(Integer.valueOf(uid)) + user.setName(name) + user.setEmail(s"[email protected]") + user.setPassword("password") + user + } + + private def grantAccess(uid: Int, privilege: PrivilegeEnum): Unit = { + val access = new WorkflowUserAccess + access.setUid(Integer.valueOf(uid)) + access.setWid(Integer.valueOf(testWorkflowWid)) + access.setPrivilege(privilege) + workflowUserAccessDao.insert(access) + } + + private def session(user: User): SessionUser = new SessionUser(user) + private def createWorkflowContent(value: String): String = { val jsonNode = objectMapper.createObjectNode() jsonNode.put("value", value) @@ -106,6 +192,38 @@ class WorkflowVersionResourceSpec patch.toString } + /** A patch whose only op is an `add`, i.e. a semantically important change. */ + private def createAddDiff(): String = { + val before = objectMapper.createObjectNode() + val after = objectMapper.createObjectNode() + after.put("addedKey", "x") + com.flipkart.zjsonpatch.JsonDiff.asJson(before, after).toString + } + + private def seedVersion(content: String, minutesAgo: Int): Unit = { + val version = new WorkflowVersion + version.setWid(Integer.valueOf(testWorkflowWid)) + version.setContent(content) + version.setCreationTime( + new Timestamp(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(minutesAgo)) + ) + workflowVersionDao.insert(version) + } + + /** The vids of the test workflow's versions, ascending (creation order). */ + private def versionVids(): List[Integer] = { + getDSLContext + .select(Tables.WORKFLOW_VERSION.VID) + .from(Tables.WORKFLOW_VERSION) + .where(Tables.WORKFLOW_VERSION.WID.eq(Integer.valueOf(testWorkflowWid))) + .orderBy(Tables.WORKFLOW_VERSION.VID.asc()) + .fetchInto(classOf[Integer]) + .asScala + .toList + } + + private def versionCount(): Int = versionVids().size + "WorkflowVersionResource" should "return versions in descending order from fetchSubsequentVersions and apply patches correctly" in { var currentContent = "initial" for (i <- 1 to 10) { @@ -192,4 +310,148 @@ class WorkflowVersionResourceSpec s"Workflow content should be 'version_$midVersionId' but was '$midVersionContent'" ) } + + it should "reconstruct an older content by applying a stored delta (applyPatch, no DB)" in { + val base = new Workflow + base.setWid(Integer.valueOf(testWorkflowWid)) + base.setContent(createWorkflowContent("new")) + + val delta = new WorkflowVersion + // a delta that turns "new" back into "old" + delta.setContent(createVersionDiff("new", "old")) + delta.setCreationTime(new Timestamp(System.currentTimeMillis())) + + val result = WorkflowVersionResource.applyPatch(List(delta), base) + + objectMapper.readTree(result.getContent).get("value").asText() shouldBe "old" + } + + "isSnapshotInRangeUnimportant" should "return true when the bounds are equal" in { + WorkflowVersionResource.isSnapshotInRangeUnimportant(7, 7, testWorkflowWid) shouldBe true + } + + it should "return true when every delta in range is positional-only (replace)" in { + seedVersion(createVersionDiff("a", "b"), minutesAgo = 2) + seedVersion(createVersionDiff("b", "c"), minutesAgo = 1) + val vids = versionVids() + + WorkflowVersionResource.isSnapshotInRangeUnimportant( + vids.head, + vids.last, + testWorkflowWid + ) shouldBe true + } + + it should "return false when a delta in range is an important (non-replace) change" in { + seedVersion(createVersionDiff("a", "b"), minutesAgo = 2) + seedVersion(createAddDiff(), minutesAgo = 1) + val vids = versionVids() + + WorkflowVersionResource.isSnapshotInRangeUnimportant( + vids.head, + vids.last, + testWorkflowWid + ) shouldBe false + } + + "insertNewVersion" should "persist a retrievable version with the given content" in { + val inserted = WorkflowVersionResource.insertNewVersion(testWorkflowWid, "[]") + + val fetched = workflowVersionDao.fetchOneByVid(inserted.getVid) + fetched should not be null + fetched.getWid shouldBe Integer.valueOf(testWorkflowWid) + fetched.getContent shouldBe "[]" + } + + "getLatestVersion" should "return the highest vid when versions exist" in { + seedVersion(createVersionDiff("a", "b"), minutesAgo = 2) + seedVersion(createVersionDiff("b", "c"), minutesAgo = 1) + val vids = versionVids() + + WorkflowVersionResource.getLatestVersion(testWorkflowWid) shouldBe vids.max + } + + it should "create and return a version when the workflow has none yet" in { + versionCount() shouldBe 0 + + val vid = WorkflowVersionResource.getLatestVersion(testWorkflowWid) + + versionCount() shouldBe 1 + workflowVersionDao.fetchOneByVid(vid) should not be null + } + + "insertVersion" should "always create a version for a new workflow" in { + WorkflowVersionResource.insertVersion(testWorkflow, insertingNewWorkflow = true) + versionCount() shouldBe 1 + } + + it should "create a version only when the content actually changed" in { + // unchanged: the passed content equals what is already stored -> empty patch -> no version + WorkflowVersionResource.insertVersion(testWorkflow, insertingNewWorkflow = false) + versionCount() shouldBe 0 + + // changed: a different content produces a non-empty patch -> one version + val edited = new Workflow + edited.setWid(Integer.valueOf(testWorkflowWid)) + edited.setContent(createWorkflowContent("changed")) + WorkflowVersionResource.insertVersion(edited, insertingNewWorkflow = false) + versionCount() shouldBe 1 + } + + "retrieveVersionsOfWorkflow" should "return the importance-encoded versions for a user with access" in { + seedVersion(createVersionDiff("a", "b"), minutesAgo = 2) + seedVersion(createVersionDiff("b", "c"), minutesAgo = 1) + + val versions = resource.retrieveVersionsOfWorkflow(testWorkflowWid, session(owner)) + + versions should have size 2 + // the latest version is always marked important + versions.head.importance shouldBe true + } + + it should "return an empty list for a user without read access" in { + seedVersion(createVersionDiff("a", "b"), minutesAgo = 1) + + resource.retrieveVersionsOfWorkflow(testWorkflowWid, session(stranger)) shouldBe empty + } + + "retrieveWorkflowVersion" should "reconstruct the workflow at a version for a user with access" in { + // an empty delta leaves the content unchanged, so the reconstructed content + // is exactly the workflow's current content + val version = WorkflowVersionResource.insertNewVersion(testWorkflowWid, "[]") + + val result = resource.retrieveWorkflowVersion(testWorkflowWid, version.getVid, session(owner)) + + result.getContent shouldBe testWorkflow.getContent + } + + it should "throw ForbiddenException for a user without read access" in { + val version = WorkflowVersionResource.insertNewVersion(testWorkflowWid, "[]") + + assertThrows[ForbiddenException] { + resource.retrieveWorkflowVersion(testWorkflowWid, version.getVid, session(stranger)) + } + } + + "cloneVersion" should "create a new workflow from an existing version" in { + // cloneVersion re-ids operators, so the version must reconstruct to a real + // workflow document; an empty delta keeps the workflow's own content. + val workflowContent = + """{"operators":[{"operatorID":"CSVFileScan-operator-a","operatorType":"CSVFileScan"}],"links":[]}""" + testWorkflow.setContent(workflowContent) + workflowDao.update(testWorkflow) + val version = WorkflowVersionResource.insertNewVersion(testWorkflowWid, "[]") + + val newWid = resource.cloneVersion( + version.getVid, + session(owner), + Map("displayedVersionId" -> 1).asJava + ) + + newWid should not be null + newWid should not be Integer.valueOf(testWorkflowWid) + val cloned = workflowDao.fetchOneByWid(newWid) + cloned should not be null + cloned.getName should include("_copy") + } }
