You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/22 22:37:46 UTC
(texera) 01/01: test(amber): cover the PVE websocket endpoint and the missing-interpreter guard (#7847)
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-7847-0d07524a5ecba3fe0f295ed404eafa4051321eb7
in repository https://gitbox.apache.org/repos/asf/texera.git
commit ce0510d32cdf654184a6d3d27ce9da11e2bf4bc6
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sat Aug 22 22:36:04 2026 +0000
test(amber): cover the PVE websocket endpoint and the missing-interpreter guard (#7847)
### What changes were proposed in this PR?
A new spec for `PveWebsocketResource.scala`, which had none, plus one
test for `PveManager.deletePackages`' missing-interpreter guard. 43 → 47
tests.
| File | Before | After |
|---|---|---|
| `PveWebsocketResource.scala` | 0/9 lines (0%) | **9/9 (100%)** |
**That 9-line figure is the only gain this PR can measure locally, and
the PveManager half is CI-only.** Measured before *and* after,
`PveManager.scala` is bit-for-bit unchanged locally: 170/228 lines,
154/228 branch-complete, 40/140 branches. The reason is that local
Windows coverage of that file is the mirror image of CI's — the
interpreter is never found here, so `PveResourceSpec`'s 6 pre-existing
failures already take both guards' true arms incidentally. On Linux
those are precisely the missed arms. On CI the bundle adds 4
fully-covered lines (453, 454, 525, 527) and completes 2 branch arms
(452, 524); since Codecov counts a line with any missed arm as missed,
those two arms are themselves 2 more fully-covered lines, so **15 lines
on Codecov's metric, 13 of which go missed → covered**.
Line 526 stays partial permanently: `logger.error(msg)` expands to
LazyLogging's one-sided `if (isErrorEnabled)`.
Attribution was measured per spec with suite-name filters, not inferred.
`PveWebsocketResourceSpec` owns all 9 lines of the websocket file *and*
`PveManager` 453/454 with arm 452; the `deletePackages` test owns
521/522/525/527 with arm 524. The two owned sets do not overlap.
### What adversarial review changed
The first draft reported 8 mutations and 0 survivors. Re-derived
independently, it was 14 mutations with **4 survivors**, and two of the
four were real defects rather than equivalent mutants:
- **Setting the pump's done flag to `false` passed every test.** That is
not equivalent: with the flag never set, the pump re-enters
`queue.take()` after `session.close()` and parks forever, because
nothing further is ever enqueued — leaking one global-EC thread per
websocket connection. The blind spot was the fixture, which stubbed
`isOpen` as always-true and `close()` as a counter, so neither
termination condition was observable. Now the fixture counts guard
evaluations: the guard reads the session exactly n times for an n-line
sentinel-terminated stream, because the (n+1)th pass short-circuits on
the flag, so a third read can only happen under the mutant. All three
tests now kill it.
- **The catch arm's `[ERR] ${e.getMessage}` could be replaced by a
constant and survive**, and the `finally` sentinel was only ever
exercised on the happy path. A new test passes an empty `packages` list,
so `.get(0)` throws inside the `Future` — before `PveManager` is entered
at all, needing no fake runner. It pins the catch arm *and* makes "the
client always gets a sentinel even when the action blows up" a proven
claim rather than a happy-path inference.
One new test was **dropped as redundant**, proven by measurement rather
than argument: `PveWebsocketResourceSpec` alone already covers the whole
line set the dropped `installUserPackages` test owned, and under the
mutation that deletes the guard's `return`, `PveResourceSpec`'s failure
count *drops* from 6 to 4 — it contributes nothing to that kill.
A fixture literal was also deleted: `"packages" ->
["colorama==0.4.6","idna"]` is unobservable, since the guard returns
before the list is read. Two version-pinned strings implying a contract
that no test constrains.
### Verification
14 mutations, **10 killed, 4 survivors**, every survivor listed. The two
repaired survivors were each confirmed *surviving on the unrepaired
bundle first*, then confirmed killed after — not assumed.
| Mutation | Killed by |
|---|---|
| pump's `done = true` → `done = false` | all 3 websocket tests, on the
guard-evaluation assertion |
| catch arm message → a constant | reports the action's own exception
and still sends the sentinel |
| `finally` sentinel `__DONE__` → `__FINISHED__` | all 3 websocket
tests, including the new failure-path one |
| `if (line == "__DONE__")` → `!=` | all 3 websocket tests |
| unknown-action message → a constant | reports an unknown action and
closes on the sentinel |
| pveName argument → a literal | routes action=install with the cuid and
pveName from the handshake |
| `cuid` parse → `0` | same test, distinct assertion |
| `deletePackages` guard message → a constant | reports a missing
interpreter and stops before the system-package check |
| `deletePackages` guard `return List(msg)` → empty list | same test,
plus an existing `PveResource` case |
| `installUserPackages` guard `return` deleted | the websocket install
test **alone** |
**The four survivors, stated plainly.** All four are mutant-surviving
gaps, not merely uncovered paths:
- `while (!done && session.isOpen)` → `while (!done)` survives: every
session in this spec is open for the whole exchange, so that conjunct
never decides anything. The pump's guard is pinned at **entry only**.
Killing it needs a closed-session test whose only available assertion is
"nothing was sent", which against a `Future` means a sleep with no event
to wait on. (The new guard-evaluation counter does not accidentally
cover it — under this mutant `isOpen` is never called at all.)
- The missing-`action` default can be replaced with garbage and survive:
all three tests supply `action` explicitly. Exercising the default means
driving `action=create`, which goes straight to the real process runner
— a real venv plus `pip install -r amber/requirements.txt`.
- Replacing the parsed package list with `Nil`, and inverting
`filter(_.nonEmpty)`, both survive — the whole 8-line parse chain is
unobservable behind the guard's early return.
### On the file reaching 100%
Worth qualifying rather than leaving as a headline. Those 9 tracked
lines are the straight-line handshake prologue. JaCoCo emits **no BRANCH
counter element** for this file: the class element contains exactly
`onOpen` and `<init>`, with no `$anonfun$onOpen$*` lambdas, because
every conditional sits inside a `Future { … }` body that the
`SyntheticFilter` drops. Those conditionals are asserted but not
counted, and only 2 of the 4 things `action` can dispatch to are
asserted at all.
### Reported, not pinned
`onOpen` parses `cuid` **outside** the `Future`, so a handshake missing
`cuid` or `pveName` throws `NumberFormatException`/NPE straight out of
`onOpen` and the client is told nothing — the catch arm inside the
`Future` cannot see it. The new failure-path test deliberately keeps
`cuid` valid and throws inside the `Future`, so it pins the catch arm
without cementing that behaviour.
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7846
### How was this PR tested?
```
sbt "WorkflowExecutionService/testOnly org.apache.texera.web.resource.pythonvirtualenvironment.PveResourceSpec org.apache.texera.web.resource.pythonvirtualenvironment.PveWebsocketResourceSpec"
```
```
[info] Tests: succeeded 41, failed 6, canceled 1, ignored 0, pending 0
```
`PveWebsocketResourceSpec` is 3/3 green across 5 runs. The 6 failures
are pre-existing on Windows and untouched by this change — they are the
`Scripts/python.exe` family, identical by name before and after, and the
pre-bundle baseline on this machine is 43 run / 37 succeeded / 6 failed
/ 1 canceled. The new guard-evaluation assertion carries a bounded 500
ms wait per test, and it is not a race: on correct code the third guard
pass short-circuits and never reads the session, so the latch can only
be counted down by a mutant.
`Test/scalafmtCheck` and `Test/scalafix --check` both pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <xi...@uci.edu>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../pythonvirtualenvironment/PveResourceSpec.scala | 25 +++
.../PveWebsocketResourceSpec.scala | 243 +++++++++++++++++++++
2 files changed, 268 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 7a63e31165..926aa87ada 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
@@ -614,4 +614,29 @@ class PveResourceSpec
PveManager.getPythonBin(testCuid, testPveName) shouldBe Some(python.toAbsolutePath.normalize())
}
+ /*
+ * `deletePackages`' missing-interpreter guard. Everything above reaches it only from the
+ * other side, with a venv already fabricated.
+ *
+ * The guard answers before the method reads `systemPackageNames`, which is what makes it
+ * safe to drive here: `systemPackages` is a lazy val resolved once per JVM through a
+ * `pip freeze`, so a test that forced it with the wrong runner installed would fix the
+ * system set for every suite that follows. This test sets no `runProcessMock`
+ * expectation, so an unexpected process spawn fails it rather than escaping to real pip —
+ * which is also what keeps "stop before the system-package check" honest.
+ *
+ * `installUserPackages`' matching guard is NOT tested here: PveWebsocketResourceSpec's
+ * install test already drives it end to end (its cuid has no venv on any platform), so a
+ * copy here would only re-cover lines that spec already owns.
+ */
+ "PveManager.deletePackages" should "report a missing interpreter and stop before the system-package check" in {
+ val absent = s"$testPveName-absent"
+
+ val output = PveManager.deletePackages(testCuid, "colorama", absent)
+
+ output should have size 1
+ output.head shouldBe
+ s"[PVE][ERR] Python executable not found for PVE: ${pythonBinFor(absent).toAbsolutePath}"
+ }
+
}
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveWebsocketResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveWebsocketResourceSpec.scala
new file mode 100644
index 0000000000..ad1f48fa03
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveWebsocketResourceSpec.scala
@@ -0,0 +1,243 @@
+/*
+ * 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.web.resource.pythonvirtualenvironment
+
+import org.apache.commons.lang3.SystemUtils
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Path, Paths}
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import javax.websocket.{RemoteEndpoint, Session}
+import scala.jdk.CollectionConverters._
+
+/**
+ * Unit tests for the PVE websocket endpoint.
+ *
+ * `onOpen` is the whole class: it reads the handshake parameters, kicks off the requested
+ * action on one Future and a pump that drains the action's log queue to the socket on
+ * another. Both Futures' bodies compile to lambdas, so JaCoCo's synthetic filter drops
+ * them from the line count entirely — the 9 lines it does track here are the straight-line
+ * handshake prologue, and it records no branch counter at all for this file. So the
+ * contracts below are what these tests actually buy; the coverage number is not evidence
+ * for any of them:
+ *
+ * - the pump forwards every queued line in order, and stops on the `__DONE__` sentinel:
+ * it closes the socket and then leaves the loop rather than parking in `queue.take()`
+ * on a queue nothing will ever fill again,
+ * - the failure path is symmetric with the happy one — the catch arm reports the action's
+ * own exception message and the `finally` still emits the sentinel, and
+ * - `action` selects the branch, with the cuid and pveName parsed off the handshake
+ * handed to `PveManager` unswapped.
+ *
+ * Only two of the four things `action` can dispatch to are asserted (`install` and the
+ * unknown-action arm). Known gaps, each stated as what survives a mutation rather than as
+ * a merely uncovered path:
+ * - `action=create`, and the `create` default a missing `action` falls back to, are not
+ * exercised, because `PveManager.createNewPve` goes straight to the production process
+ * runner: the test would build a real venv and pip-install `amber/requirements.txt`.
+ * Faking the runner is not an option either — it is a JVM-global `var`, and
+ * `PveResourceSpec` captures it in its constructor, so a mock installed here can be
+ * laundered into that suite's idea of the real runner. `PveResourceSpec` owns the
+ * create path instead. Consequence: the `getOrDefault("action", List.of("create"))`
+ * default can be replaced with any garbage string and both tests still pass.
+ * - the `packages` query parameter's parse chain is unobservable, and mutations to it
+ * survive: `installUserPackages` reports the missing interpreter and returns before it
+ * ever reads the list, so replacing the parsed list with `Nil` — or inverting the
+ * `filter(_.nonEmpty)` — changes nothing any assertion can see. Pinning it needs the
+ * same fake runner.
+ * - the pump's guard is pinned at ENTRY ONLY. `while (!done && session.isOpen)` narrowed
+ * to `while (!done)` survives, because every session here is open for the whole
+ * exchange; killing it needs a test that enters the loop with an already-closed
+ * session, whose only available assertion ("nothing was sent") is a sleep against a
+ * Future.
+ */
+class PveWebsocketResourceSpec extends AnyFlatSpec with Matchers with MockFactory {
+
+ /** Where `PveManager` looks for a PVE's interpreter on this platform. */
+ private def pythonBinFor(cuid: Int, pveName: String): Path = {
+ val venv = Paths.get("/tmp/texera-pve/venvs", cuid.toString, pveName, "pve")
+ if (SystemUtils.IS_OS_WINDOWS) venv.resolve("Scripts").resolve("python.exe")
+ else venv.resolve("bin").resolve("python")
+ }
+
+ /**
+ * How many times correct code evaluates `session.isOpen`. The pump's guard is
+ * `while (!done && session.isOpen)`, so for a stream of n lines terminated by the
+ * sentinel it evaluates the session exactly n times — the (n+1)th pass short-circuits on
+ * `!done` and never touches it. Every test here drives exactly two lines (one from the
+ * action, then `__DONE__`) and asserts that length, so two is the whole budget; a test
+ * that streams a different number of lines has to revisit this.
+ */
+ private val expectedGuardEvals = 2
+
+ /**
+ * A mocked `Session` plus the three observables the endpoint produces: the text it wrote,
+ * the close that ends the exchange, and whether the pump went back to polling the session
+ * after the sentinel.
+ *
+ * The endpoint writes from a Future, so the assertions have to wait for the close before
+ * they read `sentLines` — that is also what keeps ScalaMock's end-of-test verification
+ * from racing a call still in flight on the pump thread.
+ */
+ private class Fixture(
+ val session: Session,
+ private val sent: ConcurrentLinkedQueue[String],
+ private val closed: CountDownLatch,
+ private val extraGuardEval: CountDownLatch
+ ) {
+ def awaitClose(): Unit =
+ withClue("the endpoint never closed the session: ") {
+ closed.await(30, TimeUnit.SECONDS) shouldBe true
+ }
+
+ /**
+ * An `isOpen` evaluation past the budget means the pump re-entered its guard after
+ * writing the sentinel, i.e. `done` was never set. It then blocks in `queue.take()`
+ * forever, because the action Future has already run its `finally` and nothing will
+ * ever be enqueued again — one leaked global-EC thread per websocket connection, with
+ * the socket closed and every assertion above still green. Call after `awaitClose()`.
+ */
+ def assertPumpStopped(): Unit =
+ withClue("the pump kept polling the session after the sentinel: ") {
+ extraGuardEval.await(500, TimeUnit.MILLISECONDS) shouldBe false
+ }
+
+ def sentLines: List[String] = sent.iterator().asScala.toList
+ }
+
+ private def fixture(params: (String, String)*): Fixture =
+ fixtureOf(params.map { case (key, value) => key -> List(value).asJava }.toMap.asJava)
+
+ private def fixtureOf(
+ parameterMap: java.util.Map[String, java.util.List[String]]
+ ): Fixture = {
+ val sent = new ConcurrentLinkedQueue[String]()
+ val closed = new CountDownLatch(1)
+ val extraGuardEval = new CountDownLatch(1)
+ val guardEvals = new AtomicInteger(0)
+
+ val basic = mock[RemoteEndpoint.Basic]
+ (basic
+ .sendText(_: String))
+ .expects(*)
+ .onCall { (text: String) =>
+ sent.add(text)
+ ()
+ }
+ .anyNumberOfTimes()
+
+ val session = mock[Session]
+ (() => session.getRequestParameterMap).expects().returning(parameterMap).anyNumberOfTimes()
+ (() => session.isOpen)
+ .expects()
+ .onCall { () =>
+ if (guardEvals.incrementAndGet() > expectedGuardEvals) extraGuardEval.countDown()
+ true
+ }
+ .anyNumberOfTimes()
+ (() => session.getBasicRemote).expects().returning(basic).anyNumberOfTimes()
+ (() => session.close())
+ .expects()
+ .onCall { () =>
+ closed.countDown()
+ ()
+ }
+ .anyNumberOfTimes()
+
+ new Fixture(session, sent, closed, extraGuardEval)
+ }
+
+ "PveWebsocketResource.onOpen" should "report an unknown action and close on the sentinel" in {
+ val f = fixture(
+ "cuid" -> "424242",
+ "pveName" -> "ws-unknown-env",
+ "action" -> "explode"
+ )
+
+ new PveWebsocketResource().onOpen(f.session)
+ f.awaitClose()
+
+ // Order and length both matter: the pump must forward the action's own line first and
+ // close only once it has also written the sentinel.
+ f.sentLines shouldBe List("[ERR] Unknown action: explode", "__DONE__")
+ f.assertPumpStopped()
+ }
+
+ it should "route action=install to PveManager with the cuid and pveName from the handshake" in {
+ val cuid = 313131
+ val pveName = "ws-install-env"
+
+ // No `packages` parameter on purpose: whatever this test passed would be unobservable
+ // (see the header), and the `getOrDefault` default takes over. Its absence is the honest
+ // signal — version-pinned literals here would imply a contract nothing pins.
+ val f = fixture(
+ "cuid" -> cuid.toString,
+ "pveName" -> pveName,
+ "action" -> "install"
+ )
+
+ new PveWebsocketResource().onOpen(f.session)
+ f.awaitClose()
+
+ // No venv exists for this cuid, so PveManager's missing-interpreter guard answers. That
+ // message carries the interpreter path it derived, which is the only place the parsed
+ // cuid and pveName become observable — hence two values that cannot be mistaken for
+ // each other.
+ f.sentLines should have size 2
+ f.sentLines.head shouldBe
+ s"[PVE][ERR] Python executable not found for PVE: ${pythonBinFor(cuid, pveName).toAbsolutePath}"
+ f.sentLines.last shouldBe "__DONE__"
+ f.assertPumpStopped()
+ }
+
+ it should "report the action's own exception and still send the sentinel" in {
+ // `packages` is present but EMPTY, so `getOrDefault` hands back that empty list and the
+ // `.get(0)` on it throws inside the Future. That is the cheapest reachable failure: it
+ // needs no fake process runner, and it throws before `PveManager` is entered at all, so
+ // the memoized `systemPackages` is never forced.
+ val f = fixtureOf(
+ java.util.Map.of(
+ "cuid",
+ java.util.List.of("424243"),
+ "pveName",
+ java.util.List.of("ws-boom"),
+ "action",
+ java.util.List.of("install"),
+ "packages",
+ java.util.List.of[String]()
+ )
+ )
+
+ new PveWebsocketResource().onOpen(f.session)
+ f.awaitClose()
+
+ f.sentLines should have size 2
+ // The catch arm has to report the actual exception message (not a fixed apology).
+ // IndexOutOfBoundsException message wording varies across JDKs/collections, so match loosely.
+ f.sentLines.head should (startWith("[ERR] Index") and include("0"))
+ // And the `finally` has to emit the sentinel on the failure path too, or the client waits
+ // for an end-of-stream that never arrives.
+ f.sentLines.last shouldBe "__DONE__"
+ f.assertPumpStopped()
+ }
+}