(texera) branch gh-readonly-queue/main/pr-6128-48d9e358b4a3d86b0eb8c33ae766742dcefdd44f created (now e0475e4a57)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6128-48d9e358b4a3d86b0eb8c33ae766742dcefdd44f
in repository https://gitbox.apache.org/repos/asf/texera.git


      at e0475e4a57 test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases (#6128)

This branch includes the following new commits:

     new e0475e4a57 test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases (#6128)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(texera) 01/01: test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases (#6128)

[gi...@apache.org]

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-6128-48d9e358b4a3d86b0eb8c33ae766742dcefdd44f
in repository https://gitbox.apache.org/repos/asf/texera.git

commit e0475e4a57e6b0ac58d96a252b3c12441dcd15db
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Mon Jul 6 12:25:55 2026 -0700

    test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases (#6128)
    
    ### What changes were proposed in this PR?
    
    Adds unit-test coverage for the operator-descriptor base classes in
    `common/workflow-operator`, which Codecov reports as partly uncovered:
    
    - **`LogicalOpSpec`** (new) — exercises the `LogicalOp` base directly:
    the `getPhysicalOp` `???` default (via a minimal subtype), the
    reflection-based `hashCode`/`equals`, `toString`, and the default
    `runtimeReconfiguration` `UnsupportedOperationException`.
    - **`PythonOperatorDescriptorSpec`** (new) — exercises
    `PythonOperatorDescriptor.getPhysicalOp` through in-spec subtypes: the
    code-generation exception is embedded in the generated code rather than
    thrown, a source op gets `PreferController`, and a one-to-one op gets no
    location preference.
    - **`PythonUDFSourceOpDescV2Spec`** (extended) — adds the blank
    virtual-environment-name rejection, the trimmed `pveName`, and the
    parallelizable/`suggestedWorkerNum` path when `workers > 1`.
    
    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 "WorkflowOperator/testOnly *LogicalOpSpec *PythonOperatorDescriptorSpec *PythonUDFSourceOpDescV2Spec"
    ```
    
    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])
---
 .../texera/amber/operator/LogicalOpSpec.scala      |  74 +++++++++++++++
 .../operator/PythonOperatorDescriptorSpec.scala    | 104 +++++++++++++++++++++
 .../source/PythonUDFSourceOpDescV2Spec.scala       |  27 ++++++
 .../amber/pybuilder/PythonClassgraphScanner.scala  |   5 +
 4 files changed, 210 insertions(+)

diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/LogicalOpSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/LogicalOpSpec.scala
new file mode 100644
index 0000000000..a91c50b4ad
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/LogicalOpSpec.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.operator
+
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort}
+import org.apache.texera.amber.operator.distinct.DistinctOpDesc
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class LogicalOpSpec extends AnyFlatSpec {
+
+  private val workflowId = WorkflowIdentity(1L)
+  private val executionId = ExecutionIdentity(1L)
+
+  // a minimal subtype that leaves getPhysicalOp unoverridden to hit the base `???` default
+  private class BareLogicalOp extends LogicalOp {
+    override def operatorInfo: OperatorInfo =
+      OperatorInfo(
+        "Bare",
+        "",
+        OperatorGroupConstants.UTILITY_GROUP,
+        inputPorts = List(InputPort()),
+        outputPorts = List(OutputPort())
+      )
+  }
+
+  "LogicalOp.getPhysicalOp" should "throw NotImplementedError when a subtype does not override it" in {
+    assertThrows[NotImplementedError](new BareLogicalOp().getPhysicalOp(workflowId, executionId))
+  }
+
+  "LogicalOp.hashCode/equals" should "be reflection-based over the operator's fields" in {
+    val a = new DistinctOpDesc
+    a.setOperatorId("fixed-id")
+    assert(a.hashCode == a.hashCode) // deterministic
+    val b = new DistinctOpDesc
+    b.setOperatorId("fixed-id")
+    b.operatorVersion = a.operatorVersion
+    assert(a == b && a.hashCode == b.hashCode)
+  }
+
+  "LogicalOp.toString" should "render the class name and operator id via reflection" in {
+    val op = new DistinctOpDesc
+    op.setOperatorId("distinct-1")
+    val rendered = op.toString
+    assert(rendered.contains("DistinctOpDesc"))
+    assert(rendered.contains("distinct-1"))
+  }
+
+  "LogicalOp.runtimeReconfiguration" should "throw UnsupportedOperationException by default" in {
+    val op = new DistinctOpDesc
+    val ex = intercept[UnsupportedOperationException] {
+      op.runtimeReconfiguration(workflowId, executionId, op, op)
+    }
+    assert(ex.getMessage == "operator DistinctOpDesc does not support reconfiguration")
+  }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala
new file mode 100644
index 0000000000..e80bfefc17
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.operator
+
+import org.apache.texera.amber.core.executor.OpExecWithCode
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity, PreferController}
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PythonOperatorDescriptorSpec extends AnyFlatSpec with Matchers {
+
+  private val workflowId = WorkflowIdentity(1L)
+  private val executionId = ExecutionIdentity(1L)
+
+  private class OneToOnePyOp extends PythonOperatorDescriptor {
+    override def generatePythonCode(): String = "print('hi')"
+    override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+      Map.empty
+    override def operatorInfo: OperatorInfo =
+      OperatorInfo(
+        "t",
+        "d",
+        OperatorGroupConstants.PYTHON_GROUP,
+        List(InputPort()),
+        List(OutputPort())
+      )
+  }
+
+  private class SourcePyOp extends PythonOperatorDescriptor {
+    override def asSource(): Boolean = true
+    override def parallelizable(): Boolean = true
+    override def generatePythonCode(): String = "yield None"
+    override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+      Map.empty
+    override def operatorInfo: OperatorInfo =
+      OperatorInfo("s", "d", OperatorGroupConstants.PYTHON_GROUP, List(), List(OutputPort()))
+  }
+
+  private class ThrowingPyOp extends PythonOperatorDescriptor {
+    override def generatePythonCode(): String = throw new RuntimeException("boom")
+    override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+      Map.empty
+    override def operatorInfo: OperatorInfo =
+      OperatorInfo(
+        "x",
+        "d",
+        OperatorGroupConstants.PYTHON_GROUP,
+        List(InputPort()),
+        List(OutputPort())
+      )
+  }
+
+  private def code(op: PythonOperatorDescriptor): String =
+    op.getPhysicalOp(workflowId, executionId).opExecInitInfo match {
+      case OpExecWithCode(c, _) => c
+      case other                => fail(s"expected OpExecWithCode, got $other")
+    }
+
+  "PythonOperatorDescriptor.getPhysicalOp" should
+    "embed the generation exception in the code instead of throwing" in {
+    code(new ThrowingPyOp) shouldBe "#EXCEPTION DURING CODE GENERATION: boom"
+  }
+
+  it should "build a source PhysicalOp when asSource is true" in {
+    val physical = (new SourcePyOp).getPhysicalOp(workflowId, executionId)
+    physical.locationPreference shouldBe Some(PreferController)
+    physical.opExecInitInfo match {
+      case OpExecWithCode(c, language) =>
+        c shouldBe "yield None"
+        language shouldBe "python"
+      case other => fail(s"expected OpExecWithCode, got $other")
+    }
+    physical.isSourceOperator shouldBe true
+  }
+
+  it should "build a one-to-one PhysicalOp when asSource is false" in {
+    val physical = (new OneToOnePyOp).getPhysicalOp(workflowId, executionId)
+    physical.locationPreference shouldBe None
+    physical.opExecInitInfo match {
+      case OpExecWithCode(c, _) => c shouldBe "print('hi')"
+      case other                => fail(s"expected OpExecWithCode, got $other")
+    }
+  }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
index f0c3dfa41f..c6320d775e 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
@@ -70,6 +70,33 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with Matchers {
     intercept[IllegalArgumentException] { d.getPhysicalOp(workflowId, executionId) }
   }
 
+  it should "reject a blank virtual-environment name when the default env is disabled" in {
+    val d = new PythonUDFSourceOpDescV2
+    d.code = "yield"
+    d.defaultEnv = false
+    d.envName = "   "
+    val ex = intercept[RuntimeException] { d.getPhysicalOp(workflowId, executionId) }
+    ex.getMessage shouldBe
+      "Virtual Environment name is required when not using the default Python environment."
+  }
+
+  it should "carry the trimmed virtual-environment name when the default env is disabled" in {
+    val d = new PythonUDFSourceOpDescV2
+    d.code = "yield"
+    d.defaultEnv = false
+    d.envName = "  my-venv  "
+    d.getPhysicalOp(workflowId, executionId).pveName shouldBe "my-venv"
+  }
+
+  it should "mark the op parallelizable with the requested worker count when workers > 1" in {
+    val d = new PythonUDFSourceOpDescV2
+    d.code = "yield"
+    d.workers = 3
+    val physical = d.getPhysicalOp(workflowId, executionId)
+    physical.parallelizable shouldBe true
+    physical.suggestedWorkerNum shouldBe Some(3)
+  }
+
   "PythonUDFSourceOpDescV2" should "round-trip its config fields through the polymorphic base" in {
     val d = new PythonUDFSourceOpDescV2
     d.code = "yield"
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/pybuilder/PythonClassgraphScanner.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/pybuilder/PythonClassgraphScanner.scala
index 4a0ca124f9..f6af4fd845 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/pybuilder/PythonClassgraphScanner.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/pybuilder/PythonClassgraphScanner.scala
@@ -49,6 +49,11 @@ private[amber] object PythonClassgraphScanner {
         .toSeq
         .filterNot(_.isInterface)
         .filterNot(c => Modifier.isAbstract(c.getModifiers))
+        // Skip non-static enclosed classes (inner/local/anonymous). Real production
+        // descriptors are always top-level (or static), so this loses no coverage; it
+        // keeps the scan from picking up test-only helper subclasses declared inside a
+        // spec, which can never be instantiated via a no-arg constructor.
+        .filterNot(c => c.getEnclosingClass != null && !Modifier.isStatic(c.getModifiers))
     } finally {
       scanResult.close()
     }