build: let WorkflowOperator route Python-executing tests to amber-integration

.github/workflows/build.yml

@@ -620,6 +620,8 @@ jobs:
         # specs. The Java @TagAnnotation makes the marker visible to
         # ScalaTest's reflection, so `-n TAG` correctly narrows the
         # run.
+        # WorkflowOperator/test is here for the same reason, with its own tag:
+        # its Python-dependent tests need packages the `amber` job lacks.
         #
         # scalafmtCheckAll + scalafixAll --check are run here as well
         # because an integration-only PR fires only the
@@ -646,6 +648,7 @@ jobs:
         run: |
           sbt scalafmtCheckAll \
               "scalafixAll --check" \
+              "WorkflowOperator/test" \
               "WorkflowExecutionService/test"
       - name: Build amber dist for the texera-web boot test
         # Boot smoke test for texera-web (mirrors platform-integration for the

amber/build.sbt

@@ -69,19 +69,13 @@ Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "test" / "int
 // scalafix still cover it and `sbt Test/runMain` can invoke benches.
 Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "bench" / "scala"
 
-// Test-filter switch driven by the AMBER_TEST_FILTER env var so the
-// amber and amber-integration CI jobs select disjoint subsets without
-// each invocation having to embed a `set Tests.Argument(...)` prefix.
-//   skip-integration : exclude @IntegrationTest-tagged specs (amber job)
-//   integration-only : include only @IntegrationTest-tagged specs (amber-integration job)
-//   (unset)          : run everything (default for local sbt)
-Test / testOptions ++= (sys.env.get("AMBER_TEST_FILTER") match {
-  case Some("skip-integration") =>
-    Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", "org.apache.texera.amber.tags.IntegrationTest"))
-  case Some("integration-only") =>
-    Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", "org.apache.texera.amber.tags.IntegrationTest"))
-  case _ => Nil
-})
+// Lets the amber and amber-integration CI jobs select disjoint subsets without
+// each invocation having to embed a `set Tests.Argument(...)` prefix. See
+// project/TestFilters.scala.
+Test / testOptions ++= TestFilters.integrationSplit(
+  envVar = "AMBER_TEST_FILTER",
+  tag = "org.apache.texera.amber.tags.IntegrationTest"
+)
 
 // Excluding some proto files:
 PB.generate / excludeFilter := "scalapb.proto"

common/workflow-operator/build.sbt

@@ -35,6 +35,14 @@ ThisBuild / conflictManager := ConflictManager.latestRevision
 // Restrict parallel execution of tests to avoid conflicts
 Global / concurrentRestrictions += Tags.limit(Tags.Test, 1)
 
+// A test needing more than a bare Python interpreter is tagged, so the amber job
+// excludes it and amber-integration, which installs operator-requirements.txt,
+// runs it. The amber job already sets this env var on the step that invokes
+// WorkflowOperator/jacoco, so no workflow change is needed for the exclusion.
+Test / testOptions ++= TestFilters.integrationSplit(
+  envVar = "AMBER_TEST_FILTER",
+  tag = "org.apache.texera.amber.operator.tags.IntegrationTest"
+)
 
 /////////////////////////////////////////////////////////////////////////////
 // Compiler Options

common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java

@@ -0,0 +1,51 @@
+/*
+ * 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.tags;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.scalatest.TagAnnotation;
+
+/**
+ * Marks a test in this module as needing more than a bare Python interpreter —
+ * pandas or plotly, which the {@code amber} job does not install. See the
+ * AMBER_TEST_FILTER block in {@code common/workflow-operator/build.sbt} for how
+ * it routes to {@code amber-integration}.
+ *
+ * <p>Apply it to a whole spec as an annotation, or to a single case — {@code
+ * test(name, Tag(classOf[IntegrationTest].getName))} in a FunSuite,
+ * {@code taggedAs} in a FlatSpec — so that a spec's cheaper assertions stay in
+ * the unit job and its coverage report.
+ *
+ * <p>amber's own tag lives in {@code amber/src/test/integration} and is not
+ * reachable here, since amber depends on this module rather than the reverse.
+ *
+ * <p>Java, not Scala: ScalaTest finds tag annotations by
+ * {@code java.lang.annotation} reflection, and a Scala {@code StaticAnnotation}
+ * produces no JVM annotation interface for {@code @TagAnnotation} to mark.
+ */
+@TagAnnotation
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.TYPE})
+public @interface IntegrationTest {
+}

common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala

@@ -21,8 +21,10 @@ package org.apache.texera.amber.util
 
 import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.operator.PythonOperatorDescriptor
+import org.apache.texera.amber.operator.tags.IntegrationTest
 import org.apache.texera.amber.pybuilder.PythonReflectionTextUtils.truncateBlock
 import org.apache.texera.amber.pybuilder.PythonReflectionUtils
+import org.scalatest.Tag
 import org.scalatest.funsuite.AnyFunSuite
 
 import java.nio.charset.StandardCharsets
@@ -263,4 +265,29 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite {
     }
   }
 
+  /** py_compile above only parses the emitted code; running it needs the packages
+    * it imports. Tagged, so only amber-integration — the job that installs them —
+    * runs this. There a missing package is a defect; elsewhere it is a local-setup
+    * fact, so cancel rather than fail.
+    */
+  test(
+    "the Python interpreter operator templates run in should import pandas and plotly",
+    Tag(classOf[IntegrationTest].getName)
+  ) {
+    val provisioned = sys.env.get("AMBER_TEST_FILTER").contains("integration-only")
+    def unavailable(message: String): Nothing =
+      if (provisioned) fail(message) else cancel(message)
+
+    val python = loadPythonExeFromUdfConf().getOrElse(unavailable("no runnable python"))
+    val imported = Try {
+      val process = new ProcessBuilder(python, "-c", "import pandas, plotly")
+        .redirectErrorStream(true)
+        .start()
+      process.waitFor(60, TimeUnit.SECONDS) && process.exitValue() == 0
+    }
+    if (!imported.getOrElse(false)) {
+      unavailable(s"'$python' cannot import pandas and plotly")
+    }
+  }
+
 }

project/TestFilters.scala

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+import sbt._
+
+/**
+ * Selects a module's tagged tests for the fast-unit job or the integration job:
+ * skip-integration excludes them, integration-only runs only them, unset runs
+ * everything. Shared because the mapping is identical in every module, while the
+ * env var and the tag are not — the tag annotation has to live somewhere the
+ * module's own Test config can see.
+ */
+object TestFilters {
+
+  /** @param integrationOnlyExtra further ScalaTest args for the integration side,
+   *                              e.g. "-P4" to bound its pool when a suite forks
+   *                              a process per test.
+   */
+  def integrationSplit(
+      envVar: String,
+      tag: String,
+      integrationOnlyExtra: Seq[String] = Seq.empty
+  ): Seq[TestOption] =
+    sys.env.get(envVar) match {
+      case Some("skip-integration") =>
+        Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", tag))
+      case Some("integration-only") =>
+        Seq(Tests.Argument(TestFrameworks.ScalaTest, Seq("-n", tag) ++ integrationOnlyExtra: _*))
+      case _ => Nil
+    }
+}