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 363537e025 test(amber, workflow-operator): take the untaken branch
arms in PveManager and SQLSourceOpExec (#7699)
363537e025 is described below
commit 363537e0251182472f540079d262599f6cbb5240
Author: Meng Wang <[email protected]>
AuthorDate: Sun Aug 16 01:51:42 2026 +0000
test(amber, workflow-operator): take the untaken branch arms in PveManager
and SQLSourceOpExec (#7699)
### What changes were proposed in this PR?
Takes the untaken side of the conditionals in the two files #7698 lists.
12 new tests.
**PveManager** — the two pure guards. The create/install flows above
reach them
incidentally; these decide each conjunct directly.
- `isValidPveName`: null, a name over 128 characters (with 128 itself
pinned as
allowed), characters outside the safe set, the empty string, and a valid
name — so
each of the three `&&` operands decides the result at least once.
- `getPythonBin`: a name outside the safe set (rejected before any disk
access), an
interpreter that has not been created, one that exists but is not
executable, and one
that exists and is executable.
**SQLSourceOpExec** — the result iterator and the keyword binding,
driven against a
mocked JDBC chain.
- The iterator over a multi-row result set: one tuple per row, a second
`hasNext` that
does not consume the cached tuple, and exhaustion once the rows run out
and no further
query is available; plus a query that returns no rows at all.
- The keyword guard `keywordSearch && keywordSearchByColumn != null &&
keywords != null`
in all four shapes, asserting the bind happens only when all three hold.
The spec's existing `TestSQLSourceOpExec` already overrides
`establishConn()`, so the
mocked `Connection`/`PreparedStatement`/`ResultSet` go in through that
seam rather than by
registering a stub `java.sql.Driver` with the global `DriverManager` as
the issue
originally suggested — same hermetic result, no global state.
No production code was changed.
### One guard that cannot be reached
`getPythonBin`'s `if (!resolved.startsWith(root)) return None` is
unreachable as written:
the name must match `^[A-Za-z0-9._-]+$`, so it is a single path segment
with no separator,
and `..` climbs at most to the root itself before `pve/bin/python` is
appended — the result
always starts with the root. The test named for the traversal attempt
therefore covers the
*pattern* guard on the line above it, which is what actually rejects
such a name.
### Any related issues, documentation, discussions?
Closes #7698.
The issue originally also listed `DPThread`; it has been removed from
the issue, since
every arm it named lives inside `runDPThreadMainLogic()` —
`private[this]`, so unreachable
from a test — and `start()` constructs
`Executors.newSingleThreadExecutor` inline with no
seam to inject a same-thread executor. The one synchronously reachable
item there,
`handleActorCommand`'s two arms, is already covered by `DPThreadSpec`.
Covering the rest
needs a production seam and belongs in its own refactor issue.
### How was this PR tested?
`sbt "WorkflowOperator/testOnly *SQLSourceOpExecSpec"` — 64 pass;
`sbt "WorkflowExecutionService/testOnly *PveResourceSpec"` — 44 pass.
The failure path was
verified by breaking one assertion in each spec (red, non-zero exit) and
restoring them.
`Test/scalafmtCheck` and `Test/scalafix --check` are clean on both
modules.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../pythonvirtualenvironment/PveResourceSpec.scala | 65 ++++++++++++++++
.../operator/source/sql/SQLSourceOpExecSpec.scala | 91 ++++++++++++++++++++++
2 files changed, 156 insertions(+)
diff --git
a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
index 1934ea0d5f..7a63e31165 100644
---
a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
@@ -25,6 +25,7 @@ import
org.apache.texera.dao.jooq.generated.Tables.VIRTUAL_ENVIRONMENTS
import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
import org.apache.texera.dao.jooq.generated.tables.pojos.User
import
org.apache.texera.web.resource.pythonvirtualenvironment.PveResource.SavePvePayload
+import org.apache.commons.lang3.SystemUtils
import org.scalamock.scalatest.MockFactory
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.flatspec.AnyFlatSpec
@@ -135,6 +136,13 @@ class PveResourceSpec
PveManager.deleteEnvironments(testCuid)
}
+ /** Where PveManager looks for a venv's interpreter on this platform. */
+ private def pythonBinFor(pveName: String): Path = {
+ val venv = testRoot.resolve(pveName).resolve("pve")
+ if (SystemUtils.IS_OS_WINDOWS)
venv.resolve("Scripts").resolve("python.exe")
+ else venv.resolve("bin").resolve("python")
+ }
+
private def queueText(): String = {
queue.iterator().asScala.toList.mkString("\n")
}
@@ -549,4 +557,61 @@ class PveResourceSpec
new PveResource().listPves(sessionUser).asScala shouldBe empty
}
+ /*
+ * PveManager's two pure guards. Everything above reaches them incidentally
through the
+ * create/install flows; these take each conjunct's untaken side directly,
which is what the
+ * partially-covered branch arms on this file are.
+ */
+ "PveManager.isValidPveName" should "reject a null name" in {
+ PveManager.isValidPveName(null) shouldBe false
+ }
+
+ it should "reject a name longer than 128 characters" in {
+ PveManager.isValidPveName("a" * 129) shouldBe false
+ // The boundary itself is allowed.
+ PveManager.isValidPveName("a" * 128) shouldBe true
+ }
+
+ it should "reject a name with characters outside the safe set" in {
+ PveManager.isValidPveName("has space") shouldBe false
+ PveManager.isValidPveName("has/slash") shouldBe false
+ PveManager.isValidPveName("") shouldBe false
+ }
+
+ it should "accept a name of safe characters" in {
+ PveManager.isValidPveName("env-1.2_3") shouldBe true
+ }
+
+ "PveManager.getPythonBin" should "refuse a name outside the safe set without
touching the disk" in {
+ PveManager.getPythonBin(testCuid, "../escape") shouldBe None
+ }
+
+ it should "return nothing when the interpreter has not been created" in {
+ PveManager.getPythonBin(testCuid, testPveName) shouldBe None
+ }
+
+ it should "return nothing when the interpreter exists but is not executable"
in {
+ val python = pythonBinFor(testPveName)
+ Files.createDirectories(python.getParent)
+ Files.write(python, Array.emptyByteArray)
+ python.toFile.setExecutable(false)
+ // Clearing the bit is not something every filesystem can represent
(Windows ACLs, a
+ // root user, some mount options). Assert the state this test needs and
cancel rather
+ // than fail where the platform cannot produce it.
+ assume(!Files.isExecutable(python), "filesystem cannot represent a
non-executable file")
+
+ PveManager.getPythonBin(testCuid, testPveName) shouldBe None
+ }
+
+ it should "return the interpreter once it exists and is executable" in {
+ val python = pythonBinFor(testPveName)
+ Files.createDirectories(python.getParent)
+ Files.write(python, Array.emptyByteArray)
+ python.toFile.setExecutable(true)
+ // Likewise for the other direction: a noexec mount would keep the bit off.
+ assume(Files.isExecutable(python), "filesystem cannot represent an
executable file")
+
+ PveManager.getPythonBin(testCuid, testPveName) shouldBe
Some(python.toAbsolutePath.normalize())
+ }
+
}
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
index a54b58ede6..7d549b2c18 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
@@ -454,4 +454,95 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with
Matchers with MockFactory {
it should "tolerate being closed without a connection" in {
noException should be thrownBy new TestSQLSourceOpExec(descJson()).close()
}
+
+ /*
+ * The result iterator and the keyword binding. Everything above builds SQL
strings; these
+ * drive the executor's Iterator against a mocked JDBC chain, which is what
the untaken
+ * arms on hasNext/next and the three-way keyword guard need.
+ */
+
+ /** A connection answering one query with the given (id, name) rows, then
exhausting. */
+ private def rowsConn(rows: Seq[(Any, Any)]): Connection = {
+ val conn = mock[Connection]
+ val statement = mock[PreparedStatement]
+ val resultSet = mock[ResultSet]
+ (conn.prepareStatement(_: String)).expects(*).returning(statement)
+ (statement.executeQuery: () => ResultSet).expects().returning(resultSet)
+ inSequence {
+ rows.foreach {
+ case (id, name) =>
+ (resultSet.next _).expects().returning(true)
+ (resultSet.getObject(_: String)).expects("id").returning(id)
+ (resultSet.getObject(_: String)).expects("name").returning(name)
+ }
+ (resultSet.next _).expects().returning(false)
+ }
+ (resultSet.close _).expects()
+ (statement.close _).expects()
+ conn
+ }
+
+ private def openedPlain(
+ conn: Connection,
+ descriptor: String = descJson()
+ ): TestSQLSourceOpExec = {
+ val exec = new TestSQLSourceOpExec(descriptor, conn, execSchema =
rowSchema)
+ exec.open()
+ exec
+ }
+
+ it should "yield one tuple per row and then report exhaustion" in {
+ val exec = openedPlain(rowsConn(Seq((1, "a"), (2, "b"))))
+ val it = exec.produceTuple()
+
+ it.hasNext shouldBe true
+ // A second hasNext must not consume the cached tuple.
+ it.hasNext shouldBe true
+ it.next().asInstanceOf[Tuple].getField[Integer]("id") shouldBe 1
+ it.hasNext shouldBe true
+ it.next().asInstanceOf[Tuple].getField[Integer]("id") shouldBe 2
+
+ // The result set is drained and no further query is available.
+ it.hasNext shouldBe false
+ }
+
+ it should "report exhaustion immediately for a query that returns no rows"
in {
+ val exec = openedPlain(rowsConn(Seq.empty))
+
+ exec.produceTuple().hasNext shouldBe false
+ }
+
+ it should "bind the keyword only when the search is enabled and both column
and keywords are set" in {
+ def bindsKeyword(
+ enabled: Boolean,
+ column: Option[String],
+ keywords: Option[String],
+ expectBinding: Boolean
+ ): Unit = {
+ val conn = mock[Connection]
+ val statement = mock[PreparedStatement]
+ val resultSet = mock[ResultSet]
+ (conn.prepareStatement(_: String)).expects(*).returning(statement)
+ if (expectBinding) (statement.setString _).expects(1, keywords.get)
+ else (statement.setString _).expects(*, *).never()
+ (statement.executeQuery: () => ResultSet).expects().returning(resultSet)
+ (resultSet.next _).expects().returning(false)
+ (resultSet.close _).expects()
+ (statement.close _).expects()
+
+ val descriptor = descJson { desc =>
+ desc.keywordSearch = Option(enabled)
+ desc.keywordSearchByColumn = column
+ desc.keywords = keywords
+ }
+ openedPlain(conn, descriptor).produceTuple().hasNext shouldBe false
+ }
+
+ // each conjunct decides the outcome once
+ bindsKeyword(enabled = false, Option("name"), Option("term"),
expectBinding = false)
+ bindsKeyword(enabled = true, None, Option("term"), expectBinding = false)
+ bindsKeyword(enabled = true, Option("name"), None, expectBinding = false)
+ bindsKeyword(enabled = true, Option("name"), Option("term"), expectBinding
= true)
+ }
+
}
