(texera) branch main updated: test(workflow-core): add unit tests for PhysicalFileNode, Schema, and VFSURIFactory (#6127)

[gi...@apache.org]

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 233c4deb2b test(workflow-core): add unit tests for PhysicalFileNode, Schema, and VFSURIFactory (#6127)
233c4deb2b is described below

commit 233c4deb2bd646a5e52eea042384caa28400dbf6
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Mon Jul 6 00:35:15 2026 -0700

    test(workflow-core): add unit tests for PhysicalFileNode, Schema, and VFSURIFactory (#6127)
    
    ### What changes were proposed in this PR?
    
    Adds unit-test coverage for three value/utility classes in
    `common/workflow-core` that Codecov reports as having uncovered lines:
    
    - **`PhysicalFileNodeSpec`** (new) — covers the `PhysicalFileNode` value
    class: constructor/getters, `isFile`/`isDirectory`, `addChildNode`
    success and the direct-subpath guard failure, `equals`/`hashCode` (which
    deliberately excludes `size`), and the recursive
    `getAllFileRelativePaths` traversal.
    - **`SchemaSpec`** (extended) — adds `hashCode`/`equals` cases
    (including non-`Schema` operands), both duplicate-name rejection paths
    (`add(Iterable)` and `add(Attribute)`), and a Jackson JSON round-trip.
    - **`VFSURIFactorySpec`** (extended) — adds the missing-value guard
    where a required key is the final path segment.
    
    No production code is changed; this is test-only.
    
    ### Any related issues, documentation, discussions?
    
    Coverage gaps identified from the Codecov report for `apache/texera`.
    
    ### How was this PR tested?
    
    New/extended ScalaTest specs, run locally:
    
    ```
    sbt "WorkflowCore/testOnly *PhysicalFileNodeSpec *SchemaSpec *VFSURIFactorySpec"
    ```
    
    All pass. `scalafmt` and `scalafixAll --check` are clean.
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 4.8 [1M context])
---
 .../amber/core/storage/VFSURIFactorySpec.scala     |   7 ++
 .../util/dataset/PhysicalFileNodeSpec.scala        | 110 +++++++++++++++++++++
 .../texera/amber/core/tuple/SchemaSpec.scala       |  48 +++++++++
 3 files changed, 165 insertions(+)

diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala
index 0b8ae4a19c..f08bd292d2 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala
@@ -113,4 +113,11 @@ class VFSURIFactorySpec extends AnyFlatSpec {
       VFSURIFactory.decodeURI(new URI("vfs:///wid/1/eid/2/notarealresource"))
     }
   }
+
+  it should "reject a URI where a required key is the final segment with no value" in {
+    // "wid" is present but is the last segment (index + 1 >= segments.length)
+    assertThrows[IllegalArgumentException] {
+      VFSURIFactory.decodeURI(new URI("vfs:///eid/2/wid"))
+    }
+  }
 }
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/PhysicalFileNodeSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/PhysicalFileNodeSpec.scala
new file mode 100644
index 0000000000..d12431ae40
--- /dev/null
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/PhysicalFileNodeSpec.scala
@@ -0,0 +1,110 @@
+/*
+ * 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.amber.core.storage.util.dataset
+
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.nio.file.{Files, Path}
+import scala.jdk.CollectionConverters._
+import scala.util.Using
+
+class PhysicalFileNodeSpec extends AnyFlatSpec {
+
+  private def withRepo(body: Path => Unit): Unit = {
+    val repo = Files.createTempDirectory("pfn-")
+    try body(repo)
+    finally {
+      // Files.walk returns a Stream holding a directory handle; close it so temp
+      // cleanup does not intermittently fail (notably on Windows) on a leaked handle.
+      Using.resource(Files.walk(repo)) { stream =>
+        stream
+          .sorted(java.util.Comparator.reverseOrder[Path]())
+          .forEach(p => Files.deleteIfExists(p))
+      }
+    }
+  }
+
+  "PhysicalFileNode" should "expose its paths, size, and empty children" in withRepo { repo =>
+    val file = Files.createFile(repo.resolve("a.txt"))
+    val node = new PhysicalFileNode(repo, file, 42L)
+    assert(node.getAbsolutePath == file)
+    assert(node.getRelativePath == repo.relativize(file))
+    assert(node.getRelativePath.toString == "a.txt")
+    assert(node.getSize == 42L)
+    assert(node.getChildren.isEmpty)
+  }
+
+  it should "report isFile / isDirectory from the filesystem" in withRepo { repo =>
+    val file = Files.createFile(repo.resolve("a.txt"))
+    val dir = Files.createDirectory(repo.resolve("sub"))
+    val fileNode = new PhysicalFileNode(repo, file, 0L)
+    val dirNode = new PhysicalFileNode(repo, dir, 0L)
+    assert(fileNode.isFile && !fileNode.isDirectory)
+    assert(dirNode.isDirectory && !dirNode.isFile)
+    val missing = new PhysicalFileNode(repo, repo.resolve("nope"), 0L)
+    assert(!missing.isFile && !missing.isDirectory)
+  }
+
+  it should "accept a direct child and reject a non-direct one" in withRepo { repo =>
+    val sub = Files.createDirectory(repo.resolve("sub"))
+    val parent = new PhysicalFileNode(repo, sub, 0L)
+    val child = new PhysicalFileNode(repo, sub.resolve("child.txt"), 0L)
+    parent.addChildNode(child)
+    assert(parent.getChildren.size == 1)
+    assert(parent.getChildren.contains(child))
+
+    val parentOfRepo = new PhysicalFileNode(repo, repo, 0L)
+    val deep = new PhysicalFileNode(repo, sub.resolve("deep.txt"), 0L)
+    val ex = intercept[IllegalArgumentException](parentOfRepo.addChildNode(deep))
+    assert(ex.getMessage == "Child node is not a direct subpath of the parent node")
+  }
+
+  it should "define equals and hashCode over the absolute path and children only" in withRepo {
+    repo =>
+      val file = Files.createFile(repo.resolve("a.txt"))
+      val other = Files.createFile(repo.resolve("b.txt"))
+      val node = new PhysicalFileNode(repo, file, 1L)
+      assert(node.equals(node)) // identity
+      assert(!node.equals(null))
+      assert(!node.equals("not a node"))
+      // size is excluded from equals/hashCode
+      val sameByPath = new PhysicalFileNode(repo, file, 999L)
+      assert(node == sameByPath)
+      assert(node.hashCode == sameByPath.hashCode)
+      assert(node != new PhysicalFileNode(repo, other, 1L))
+  }
+
+  "PhysicalFileNode.getAllFileRelativePaths" should "collect files recursively through directories" in withRepo {
+    repo =>
+      val sub = Files.createDirectory(repo.resolve("sub"))
+      val topFile = Files.createFile(repo.resolve("top.txt"))
+      val innerFile = Files.createFile(sub.resolve("inner.txt"))
+      val topNode = new PhysicalFileNode(repo, topFile, 0L)
+      val dirNode = new PhysicalFileNode(repo, sub, 0L)
+      dirNode.addChildNode(new PhysicalFileNode(repo, innerFile, 0L))
+      val nodes = new java.util.HashSet[PhysicalFileNode]()
+      nodes.add(topNode)
+      nodes.add(dirNode)
+      val result = PhysicalFileNode.getAllFileRelativePaths(nodes).asScala.toSet
+      assert(
+        result == Set(repo.relativize(topFile).toString, repo.relativize(innerFile).toString)
+      )
+  }
+}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/SchemaSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/SchemaSpec.scala
index 1e2d22d6a3..cf62f7ba15 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/SchemaSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/SchemaSpec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.texera.amber.core.tuple
 
+import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.flatspec.AnyFlatSpec
 
 class SchemaSpec extends AnyFlatSpec {
@@ -274,4 +275,51 @@ class SchemaSpec extends AnyFlatSpec {
       schema.toString == "Schema[Attribute[name=id, type=integer], Attribute[name=name, type=string]]"
     )
   }
+
+  it should "produce equal, deterministic hash codes for equal schemas" in {
+    val a = Schema(
+      List(new Attribute("id", AttributeType.INTEGER), new Attribute("name", AttributeType.STRING))
+    )
+    val b = Schema(
+      List(new Attribute("id", AttributeType.INTEGER), new Attribute("name", AttributeType.STRING))
+    )
+    assert(a.hashCode == a.hashCode)
+    assert(a == b && a.hashCode == b.hashCode)
+    assert(Schema().hashCode == Schema().hashCode)
+  }
+
+  it should "not be equal to a non-Schema object" in {
+    val schema = Schema(List(new Attribute("id", AttributeType.INTEGER)))
+    assert(!schema.equals("not a schema"))
+    assert(!schema.equals(42))
+    assert(!schema.equals(null))
+  }
+
+  it should "throw when add(Iterable) contains a name already in the schema" in {
+    val schema = Schema(List(new Attribute("id", AttributeType.INTEGER)))
+    val ex = intercept[RuntimeException] {
+      schema.add(List(new Attribute("id", AttributeType.STRING)))
+    }
+    assert(ex.getMessage == "Cannot add attributes with duplicate names: id")
+  }
+
+  it should "throw when add(Attribute) name already exists" in {
+    val schema = Schema(List(new Attribute("id", AttributeType.INTEGER)))
+    val ex = intercept[RuntimeException] {
+      schema.add(new Attribute("id", AttributeType.LONG))
+    }
+    assert(ex.getMessage == "Attribute name 'id' already exists in the schema")
+  }
+
+  it should "round-trip through Jackson via the @JsonCreator constructor" in {
+    val schema =
+      Schema(
+        List(
+          new Attribute("id", AttributeType.INTEGER),
+          new Attribute("name", AttributeType.STRING)
+        )
+      )
+    val restored = objectMapper.readValue(objectMapper.writeValueAsString(schema), classOf[Schema])
+    assert(restored == schema)
+  }
 }