Copilot commented on code in PR #7273:
URL: https://github.com/apache/texera/pull/7273#discussion_r3700949419
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala:
##########
@@ -19,9 +19,223 @@
package org.apache.texera.amber.core.storage.util
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.sun.net.httpserver.{HttpExchange, HttpServer}
+import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.common.tags.NonParallelTest
import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
-class LakeFSStorageClientSpec extends AnyFlatSpec {
+import java.net.{InetAddress, InetSocketAddress, URLDecoder}
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.Files
+import java.util.concurrent.{ConcurrentLinkedQueue, ExecutorService, Executors}
+import scala.jdk.CollectionConverters._
+
+/**
+ * Request/response records for the loopback stub below.
+ *
+ * Top-level (rather than nested in the suite) so that `case
StubRequest(...)` patterns are not
+ * path-dependent — the compiler cannot check the outer reference of an inner
case class at run
+ * time and warns on every match site.
+ */
+private object LakeFSStubServer {
+ final case class StubRequest(
+ method: String,
+ path: String,
+ query: Map[String, String],
+ body: String
+ )
+
+ final case class StubResponse(status: Int, body: String = "")
+}
+
+/**
+ * Spec for the parts of [[LakeFSStorageClient]] that do not need a real
lakeFS server:
+ *
+ * - pure helpers (`parsePhysicalAddress`, the `initRepo` name validation),
and
+ * - the request/response wiring, driven against a loopback stub that
speaks just enough of the
+ * lakeFS REST API for the generated SDK to be satisfied.
+ *
+ * The stub is deliberately *not* a way to test the SDK's URL templates. What
it pins is the set of
+ * choices this class makes on top of the SDK and that nothing else can
observe:
+ *
+ * - which ref each call targets — several methods hard-code the `main`
branch
+ * (`deleteObject`, `resetObjectUploadOrDeletion`, the multipart calls)
while their siblings
+ * take a caller-supplied branch (`createCommit`) or a commit hash
(`getFileSize`,
+ * `getFilePresignedUrl`, `getFileFromRepo`); mixing them up is silent
and destructive;
+ * - the `fetchAllPages` loop — cursor threading and page accumulation,
which a live server can
+ * only exercise past 1000 objects;
+ * - `retrieveVersionsOfRepository`'s newest-commit-first ordering, which a
live server hides
+ * because it already returns commits newest-first;
+ * - `completePresignedMultipartUploads`'s part sort, which a live server
absorbs silently;
+ * - which field of a stat response each getter returns.
+ *
+ * A stub is not just a convenience here: only against one can you assert
that ZERO bytes left the
+ * process (see the `initRepo` validation test). A live server can
distinguish reject-before-send
+ * from server-side reject by exception type, but not prove nothing was sent.
+ *
+ * Tagged [[NonParallelTest]] so `common/workflow-core/build.sbt` gives this
suite its own forked
+ * JVM. That is load-bearing, not cosmetic: `LakeFSStorageClient.apiClient`
is a `lazy val` that
+ * captures `StorageConfig.lakefsEndpoint` once per JVM, and
`LakeFSStorageClientMtimeSpec` points
+ * that same endpoint at a testcontainer. The two suites must never share a
JVM; both are tagged,
+ * so isolation survives either tag being dropped.
+ */
+@NonParallelTest
+class LakeFSStorageClientSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach {
+
+ //
---------------------------------------------------------------------------------------------
+ // Loopback stub
+ //
---------------------------------------------------------------------------------------------
+
+ import LakeFSStubServer._
+
+ private val requests = new ConcurrentLinkedQueue[StubRequest]()
+
+ private val notStubbed: StubRequest => StubResponse =
+ req => StubResponse(501, s"""{"message":"no stub route for ${req.method}
${req.path}"}""")
+
+ @volatile private var route: StubRequest => StubResponse = notStubbed
+
+ private var server: HttpServer = _
+ private var serverPool: ExecutorService = _
+
+ private def decode(s: String): String = URLDecoder.decode(s, UTF_8.name())
+
+ private def handle(exchange: HttpExchange): Unit = {
+ try {
+ val body = new String(exchange.getRequestBody.readAllBytes(), UTF_8)
+ val query = Option(exchange.getRequestURI.getRawQuery)
+ .filter(_.nonEmpty)
+ .map(_.split("&").toList.map { pair =>
+ pair.indexOf('=') match {
+ case -1 => decode(pair) -> ""
+ case i => decode(pair.substring(0, i)) -> decode(pair.substring(i
+ 1))
+ }
+ }.toMap)
+ .getOrElse(Map.empty[String, String])
+
+ val request =
+ StubRequest(exchange.getRequestMethod, exchange.getRequestURI.getPath,
query, body)
+ requests.add(request)
+
+ val response =
+ try route(request)
+ catch { case t: Throwable => StubResponse(500,
s"""{"message":"${t.getClass.getName}"}""") }
+
+ val bytes = response.body.getBytes(UTF_8)
+ if (bytes.isEmpty) {
+ // -1 means "no response body"; the SDK maps 204 to a null (Unit)
return.
+ exchange.sendResponseHeaders(response.status, -1L)
+ } else {
+ exchange.getResponseHeaders.set("Content-Type", "application/json")
+ exchange.sendResponseHeaders(response.status, bytes.length.toLong)
+ exchange.getResponseBody.write(bytes)
+ }
+ } finally exchange.close()
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ server = HttpServer.create(new
InetSocketAddress(InetAddress.getLoopbackAddress, 0), 0)
+ server.createContext("/", (exchange: HttpExchange) => handle(exchange))
+ serverPool = Executors.newFixedThreadPool(2)
+ server.setExecutor(serverPool)
+ server.start()
+ // Must happen before anything forces LakeFSStorageClient.apiClient (a
JVM-wide lazy val).
+ // Nothing above this line touches the client, and no other suite shares
this forked JVM.
+ StorageConfig.lakefsEndpoint =
s"http://127.0.0.1:${server.getAddress.getPort}/api/v1";
Review Comment:
The stub server binds to `InetAddress.getLoopbackAddress` but the client
endpoint is hard-coded to `http://127.0.0.1:...`. On systems where the loopback
address resolves to IPv6 (`::1`, e.g. when
`java.net.preferIPv6Addresses=true`), the server can bind only on `::1` and
connections to `127.0.0.1` will fail. Bind explicitly to `127.0.0.1` (or derive
the endpoint host from the bound address) so the server bind and client
endpoint are always consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
