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 21823145b3 test(workflow-operator): add unit tests for LogicalOp and
Python operator descriptor bases (#6128)
21823145b3 is described below
commit 21823145b328a91948fa6ec8cc24c739665099e5
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Jul 6 23:45:58 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 | 109 +++++++++++++++++++++
.../source/PythonUDFSourceOpDescV2Spec.scala | 27 +++++
.../amber/pybuilder/PythonClassgraphScanner.scala | 5 +
4 files changed, 215 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..bdeb3429b5
--- /dev/null
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala
@@ -0,0 +1,109 @@
+/*
+ * 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,
+ PreferCoordinator
+}
+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(PreferCoordinator)
+ 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()
}
