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-7699-0da794d90f0ed5462b6b12aed4238b10996a6b08 in repository https://gitbox.apache.org/repos/asf/texera.git
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) + } + }
