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 2fb1ff475b test(workflow-core): cover LakeFSStorageClient against a
local stub server (#7273)
2fb1ff475b is described below
commit 2fb1ff475bf662f1e093e10c3448368b29be28b9
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Aug 3 22:37:37 2026 -0700
test(workflow-core): cover LakeFSStorageClient against a local stub server
(#7273)
### What changes were proposed in this PR?
`LakeFSStorageClient` was the largest unclaimed gap in the repo — 83
missed, 37.9%. The existing spec only exercised `parsePhysicalAddress`;
everything else needed a lakeFS server, and the only harness for that is
the Docker/testcontainers `MtimeSpec`.
This adds **20 tests** driven by a JDK `com.sun.net.httpserver` stub on
`127.0.0.1:0` — no new dependency, no container.
`StorageConfig.lakefsEndpoint` is a `var` precisely so tests can repoint
it.
**This is not an absence-of-service test.** Nothing asserts a connection
failure or an `ApiException` caused by one, so nothing inverts if CI
ever gains a lakeFS container. Every response is an explicit canned
payload carrying exactly the fields the generated SDK marks required
(read out of the sdk-1.51.0 sources), so the fixtures cannot silently
rot into "any JSON passes".
Several behaviours are **only** observable against a stub:
| Behaviour | Why a live server can't show it |
|---|---|
| `fetchAllPages` cursor threading | `PageSize` is 1000 — a real server
needs 1001 objects for a second page |
| `retrieveVersionsOfRepository` descending sort | lakeFS already
returns commits newest-first, so the sort is a no-op; the stub returns
them out of order with the newest neither first nor last |
| `completePresignedMultipartUploads` part ordering | a live server
absorbs the order silently; the stub inspects the serialized body |
| "rejected without issuing a request" | indistinguishable from a
server-side reject |
**Both suites are now tagged `@NonParallelTest`.** This matters:
`apiClient` is a `private lazy val` (as are the seven `*Api` vals
derived from it), so the endpoint is captured **once per JVM** on first
force, and `MtimeSpec` repoints it at a container. The old pair avoided
the clash only by accident — the old spec never forced `apiClient`.
`build.sbt` already gives each tagged suite its own forked
`Tests.Group`; tagging both means the isolation survives either tag
being dropped later. Verified with `show WorkflowCore/Test/testGrouping`
rather than assumed. Tagging `MtimeSpec` is independently correct anyway
— it mutates `StorageConfig.s3Endpoint`, exactly the hazard that tag
exists for.
**The page-1 stub routes serve once.** Without that guard, a client that
stopped threading the cursor would re-match a `has_more:true` route
forever and **hang CI** rather than fail — the regression these tests
exist to catch would have become a timeout. Verified by breaking the
cursor in production: the suite now fails 3 tests in seconds instead of
spinning.
Measured with jacoco on a clean single-suite run:
```
before: 23/152 lines covered (15.1%)
after: 132/152 lines covered (86.8%)
```
Assertion strength checked by **20 production mutations**, each killed
by exactly the test that claims to guard it, no survivors — the
repo-name regex bounds, the per-repository storage namespace,
page-buffer accumulation, the commit sort, sum-vs-max, the empty-repo
short-circuit, part ordering, the physical-address on abort, reset type,
commit branch, stat field, download ref, the null guard, part count,
size-vs-mtime, `PageSize`, delete argument order, diff branch, and
`.trim`. All reverted; the production diff is empty.
Deliberately left uncovered, with a comment so they stay easy to delete:
`removeFileFromRepo` and `retrieveFileContent` have **no callers
anywhere in the repo**, `withCreateVersion`'s only caller is a
file-service test, and `stagingApi` is referenced nowhere — not even
inside its own class. `deleteRepo` is a bodyless passthrough with no
choice of its own to assert. Also skipped: `healthCheck`'s retry path,
which would burn ~3.2s sleeping and pin `HealthCheckMaxAttempts`, a
constant a harmless tuning change should be free to alter — the success
path does assert exactly one probe, which is the invariant worth
holding.
### Any related issues, documentation, discussions?
Closes #7272
### How was this PR tested?
```
sbt "WorkflowCore/testOnly
org.apache.texera.amber.core.storage.util.LakeFSStorageClientSpec"
```
```
[info] Tests: succeeded 25, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
`WorkflowCore/Test/scalafmtCheck` and `WorkflowCore/Test/scalafix
--check` both `[success]`.
One caveat worth stating plainly: **`LakeFSStorageClientMtimeSpec` could
not be run locally** — Docker Desktop would not come up on this machine,
so it aborts with "Could not find a valid Docker environment". The only
edits to that file are the `@NonParallelTest` annotation and a scaladoc
paragraph; no test body was touched. CI will be the first green run of
that suite.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../util/LakeFSStorageClientMtimeSpec.scala | 9 +
.../storage/util/LakeFSStorageClientSpec.scala | 602 ++++++++++++++++++++-
2 files changed, 610 insertions(+), 1 deletion(-)
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientMtimeSpec.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientMtimeSpec.scala
index 84d445be7f..c08ad1d6c2 100644
---
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientMtimeSpec.scala
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientMtimeSpec.scala
@@ -28,6 +28,7 @@ import com.dimafeng.testcontainers.{
}
import io.lakefs.clients.sdk.ApiException
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.testcontainers.containers.Network
@@ -47,7 +48,15 @@ import java.nio.charset.StandardCharsets
* The method is a thin LakeFS-SDK passthrough (statObject -> mtime), so it
can only be
* exercised against a real LakeFS. This spins up the same Postgres + MinIO +
LakeFS stack
* the file-service tests use, with Postgres backing the LakeFS metadata
store.
+ *
+ * Tagged [[NonParallelTest]] so `common/workflow-core/build.sbt` gives this
suite its own forked
+ * JVM. It repoints two JVM-wide singletons — `StorageConfig.s3Endpoint` and
+ * `StorageConfig.lakefsEndpoint` — at its containers, and
`LakeFSStorageClient.apiClient` is a
+ * `lazy val` that captures the endpoint on first use and never releases it.
Sharing a JVM with
+ * `LakeFSStorageClientSpec`, which points the same endpoint at a loopback
stub, would make
+ * whichever suite ran second talk to the other one's server.
*/
+@NonParallelTest
class LakeFSStorageClientMtimeSpec
extends AnyFlatSpec
with Matchers
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
index ad2bb38977..fbd4289c97 100644
---
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
+++
b/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.{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("127.0.0.1", 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";
+ }
+
+ override def afterAll(): Unit = {
+ try {
+ if (server != null) server.stop(0)
+ if (serverPool != null) serverPool.shutdownNow()
+ } finally super.afterAll()
+ }
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ requests.clear()
+ route = notStubbed
+ }
+
+ /** Installs the routes for one test; anything unmatched answers 501 and
shows up in `recorded`. */
+ private def stub(routes: PartialFunction[StubRequest, StubResponse]): Unit =
+ route = req => routes.applyOrElse(req, notStubbed)
+
+ private def recorded: List[StubRequest] = requests.asScala.toList
+
+ /**
+ * True only for the FIRST request this test makes to `path`. Page-1 routes
are guarded with it so
+ * that a broken pagination cursor produces a clean 501 -> ApiException
instead of an infinite
+ * loop: without it, a client that stops threading `after` re-matches the
`has_more:true` page-1
+ * route forever and the suite hangs rather than failing. `requests.add`
runs before the route is
+ * applied, so the current request is already counted.
+ */
+ private def firstHitOf(path: String): Boolean = recorded.count(_.path ==
path) == 1
+
+ private def onlyRequest: StubRequest = {
+ recorded should have size 1
+ recorded.head
+ }
+
+ private val mapper = new ObjectMapper()
+ private def json(body: String) = mapper.readTree(body)
+
+ //
---------------------------------------------------------------------------------------------
+ // Response fixtures. Only the fields the SDK marks required, plus whatever
a test reads back.
+ //
---------------------------------------------------------------------------------------------
+
+ private def objectStats(path: String, sizeBytes: Long, physicalAddress:
String): String =
+
s"""{"path":"$path","path_type":"object","physical_address":"$physicalAddress",
+
|"checksum":"chk","mtime":1700000000,"size_bytes":$sizeBytes}""".stripMargin
+
+ private def pagination(hasMore: Boolean, nextOffset: String, results: Int):
String =
+
s"""{"has_more":$hasMore,"next_offset":"$nextOffset","results":$results,"max_per_page":1000}"""
+
+ private def objectPage(
+ hasMore: Boolean,
+ nextOffset: String,
+ objects: (String, Long)*
+ ): String = {
+ val results = objects.map { case (p, s) => objectStats(p, s,
s"s3://bucket/$p") }.mkString(",")
+ s"""{"pagination":${pagination(hasMore, nextOffset,
objects.size)},"results":[$results]}"""
+ }
+
+ private def diffPage(hasMore: Boolean, nextOffset: String, paths: String*):
String = {
+ val results = paths.map(p =>
s"""{"type":"added","path":"$p","path_type":"object"}""")
+ s"""{"pagination":${pagination(hasMore, nextOffset, paths.size)},
+ |"results":[${results.mkString(",")}]}""".stripMargin
+ }
+
+ private def commitList(commits: (String, Long)*): String = {
+ val results = commits
+ .map {
+ case (id, createdAt) =>
+ s"""{"id":"$id","parents":[],"committer":"tester","message":"m",
+ |"creation_date":$createdAt,"meta_range_id":"mr"}""".stripMargin
+ }
+ .mkString(",")
+ s"""{"pagination":${pagination(false, "",
commits.size)},"results":[$results]}"""
+ }
+
+ private def repository(id: String): String =
+ s"""{"id":"$id","creation_date":1700000000,"default_branch":"main",
+ |"storage_namespace":"s3://bucket/$id"}""".stripMargin
+
+ private def commit(id: String): String =
+ s"""{"id":"$id","parents":[],"committer":"tester","message":"m",
+ |"creation_date":1700000000,"meta_range_id":"mr"}""".stripMargin
+
+ //
---------------------------------------------------------------------------------------------
+ // parsePhysicalAddress
+ //
---------------------------------------------------------------------------------------------
// `healthCheck` retries through the shared `RetryUtil.withBackoff`; that
contract (progression,
// give-up wrapping, interrupt fail-fast) is covered by `RetryUtilSpec` in
`common/util`.
@@ -77,4 +291,390 @@ class LakeFSStorageClientSpec extends AnyFlatSpec {
}
assert(rootPathEx.getMessage.contains("missing key/path"))
}
+
+ it should "reject a null address instead of dereferencing it" in {
+ // The `Option(address)` guard is what turns a null into an
IllegalArgumentException instead
+ // of an NPE. (Its one caller already null-guards upstream, so this pins
the documented
+ // @throws contract rather than a reachable production path.)
+ val ex = intercept[IllegalArgumentException] {
+ LakeFSStorageClient.parsePhysicalAddress(null)
+ }
+ assert(ex.getMessage.contains("empty"))
+ }
+
+ it should "trim surrounding whitespace before parsing" in {
+ // Without the leading trim the URI parse fails (a space is not a legal
URI character), so this
+ // pins the trim rather than restating the happy path.
+ assert(
+ LakeFSStorageClient.parsePhysicalAddress("
s3://my-bucket/path/to/file.csv \n") ==
+ (("my-bucket", "path/to/file.csv"))
+ )
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // initRepo
+ //
---------------------------------------------------------------------------------------------
+
+ "initRepo" should "reject malformed repository names without issuing a
request" in {
+ val invalid = Seq(
+ "ab" -> "two characters is below the 3-character minimum",
+ "a" * 64 -> "64 characters is above the 63-character maximum",
+ "-abc" -> "a leading hyphen is not allowed",
+ "Abc" -> "uppercase is not allowed",
+ "a_bc" -> "underscores are not allowed",
+ "a.bc" -> "dots are not allowed",
+ "" -> "empty is not a name"
+ )
+
+ invalid.foreach {
+ case (name, why) =>
+ val ex = intercept[IllegalArgumentException] {
+ LakeFSStorageClient.initRepo(name)
+ }
+ withClue(s"$name ($why): ") {
+ ex.getMessage should include(s"'$name'")
+ }
+ }
+
+ // Validation is client-side: a rejected name must never reach lakeFS.
Only a stub can show
+ // this — against a live server a rejected create and a never-sent create
look identical.
+ recorded shouldBe empty
+ }
+
+ it should "accept the shortest and longest legal repository names" in {
+ // Pins both ends of the {2,62}-after-first-character window: 3 and 63
characters inclusive.
+ // Without this, narrowing the quantifier by one on either side goes
unnoticed.
+ Seq("abc", "a" * 63).foreach { name =>
+ requests.clear()
+ stub {
+ case StubRequest("POST", "/api/v1/repositories", _, _) =>
+ StubResponse(201, repository(name))
+ }
+ LakeFSStorageClient.initRepo(name).getId shouldEqual name
+ json(onlyRequest.body).get("name").asText() shouldEqual name
+ }
+ }
+
+ it should "create the repository under a per-repository storage namespace on
the main branch" in {
+ val name = "texera-init-repo"
+ stub {
+ case StubRequest("POST", "/api/v1/repositories", _, _) =>
StubResponse(201, repository(name))
+ }
+
+ LakeFSStorageClient.initRepo(name)
+
+ val body = json(onlyRequest.body)
+ body.get("name").asText() shouldEqual name
+ body.get("default_branch").asText() shouldEqual "main"
+ body.get("sample_data").asBoolean() shouldEqual false
+ // Every repository gets its own prefix *inside* the shared bucket — two
repositories must not
+ // land on the same namespace, which is what the trailing "/<repo>"
segment guarantees.
+ val namespace = body.get("storage_namespace").asText()
+ namespace should startWith(
+
s"${StorageConfig.lakefsBlockStorageType}://${StorageConfig.lakefsBucketName}"
+ )
+ namespace should endWith(s"/$name")
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Pagination (fetchAllPages)
+ //
---------------------------------------------------------------------------------------------
+
+ "retrieveObjectsOfVersion" should "follow the pagination cursor and
concatenate every page" in {
+ val repo = "texera-paging"
+ val hash = "commit-abc"
+ stub {
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/ls" &&
!q.contains(
+ "after"
+ ) && firstHitOf(p) =>
+ StubResponse(200, objectPage(hasMore = true, "cursor-1", "a.txt" ->
1L, "b.txt" -> 2L))
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/ls" &&
+ q.get("after").contains("cursor-1") =>
+ StubResponse(200, objectPage(hasMore = false, "", "c.txt" -> 3L))
+ }
+
+ val objects = LakeFSStorageClient.retrieveObjectsOfVersion(repo, hash)
+
+ // Page 2 is appended to page 1, in page order — not overwritten by the
last page.
+ objects.map(_.getPath) shouldEqual List("a.txt", "b.txt", "c.txt")
+
+ recorded should have size 2
+ // The first page is requested with no cursor; the second carries the
cursor the first returned.
+ recorded.head.query.get("after") shouldBe None
+ recorded(1).query.get("after") shouldBe Some("cursor-1")
+ // Both pages ask for the class's page size, so a 110-object repository is
one round trip.
+ recorded.map(_.query.get("amount")) shouldEqual List(Some("1000"),
Some("1000"))
+ }
+
+ "retrieveUncommittedObjects" should "page through the diff of the main
branch" in {
+ val repo = "texera-uncommitted"
+ stub {
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/branches/main/diff" &&
!q.contains(
+ "after"
+ ) && firstHitOf(p) =>
+ StubResponse(200, diffPage(hasMore = true, "cursor-1",
"staged/one.bin"))
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/branches/main/diff" &&
+ q.get("after").contains("cursor-1") =>
+ StubResponse(200, diffPage(hasMore = false, "", "staged/two.bin"))
+ }
+
+ LakeFSStorageClient
+ .retrieveUncommittedObjects(repo)
+ .map(_.getPath) shouldEqual List("staged/one.bin", "staged/two.bin")
+
+ // Uncommitted work only ever lives on the write branch, so the diff is
pinned to main.
+ recorded.map(_.path).distinct shouldEqual
List(s"/api/v1/repositories/$repo/branches/main/diff")
+ recorded should have size 2
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // retrieveRepositorySize
+ //
---------------------------------------------------------------------------------------------
+
+ "retrieveRepositorySize" should "sum every page of a given commit without
reading the log" in {
+ val repo = "texera-size"
+ val hash = "commit-xyz"
+ stub {
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/ls" &&
!q.contains(
+ "after"
+ ) && firstHitOf(p) =>
+ StubResponse(200, objectPage(hasMore = true, "cursor-1", "a" -> 10L,
"b" -> 20L))
+ case StubRequest("GET", p, q, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/ls" &&
+ q.get("after").contains("cursor-1") =>
+ StubResponse(200, objectPage(hasMore = false, "", "c" -> 5L))
+ }
+
+ LakeFSStorageClient.retrieveRepositorySize(repo, hash) shouldEqual 35L
+ // An explicit hash short-circuits the commit lookup entirely.
+ recorded.exists(_.path.endsWith("/commits")) shouldBe false
+ }
+
+ it should "measure the newest commit when no commit hash is given" in {
+ val repo = "texera-size-head"
+ // Deliberately out of order, and with the newest neither first nor last
in the response, so
+ // neither `.head` nor `.last` of the raw list would pick it.
+ stub {
+ case StubRequest("GET", p, _, _) if p ==
s"/api/v1/repositories/$repo/refs/main/commits" =>
+ StubResponse(200, commitList("older" -> 100L, "newest" -> 300L,
"middle" -> 200L))
+ case StubRequest("GET", p, _, _)
+ if p == s"/api/v1/repositories/$repo/refs/newest/objects/ls" =>
+ StubResponse(200, objectPage(hasMore = false, "", "a" -> 7L, "b" ->
8L))
+ }
+
+ LakeFSStorageClient.retrieveRepositorySize(repo) shouldEqual 15L
+ // The objects listed are the ones at the newest commit, not at whichever
came back first.
+ recorded.map(_.path) shouldEqual List(
+ s"/api/v1/repositories/$repo/refs/main/commits",
+ s"/api/v1/repositories/$repo/refs/newest/objects/ls"
+ )
+ }
+
+ it should "report zero for a repository with no commits, without listing
objects" in {
+ val repo = "texera-size-empty"
+ stub {
+ case StubRequest("GET", p, _, _) if p ==
s"/api/v1/repositories/$repo/refs/main/commits" =>
+ StubResponse(200, commitList())
+ }
+
+ LakeFSStorageClient.retrieveRepositorySize(repo) shouldEqual 0L
+ // Listing at the empty ref would be a request lakeFS rejects; the guard
must stop first.
+ onlyRequest.path shouldEqual
s"/api/v1/repositories/$repo/refs/main/commits"
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Presigned multipart uploads
+ //
---------------------------------------------------------------------------------------------
+
+ "initiatePresignedMultipartUploads" should "request the given part count on
the main branch" in {
+ val repo = "texera-mpu"
+ val path = "data/big.bin"
+ stub {
+ case StubRequest("POST", p, _, _)
+ if p == s"/api/v1/repositories/$repo/branches/main/staging/pmpu" =>
+ StubResponse(201,
"""{"upload_id":"upload-1","physical_address":"s3://bucket/phys"}""")
+ }
+
+ val upload = LakeFSStorageClient.initiatePresignedMultipartUploads(repo,
path, 7)
+ upload.getUploadId shouldEqual "upload-1"
+ upload.getPhysicalAddress shouldEqual "s3://bucket/phys"
+
+ onlyRequest.query.get("parts") shouldBe Some("7")
+ onlyRequest.query.get("path") shouldBe Some(path)
+ }
+
+ "completePresignedMultipartUploads" should "send parts ordered by part
number" in {
+ val repo = "texera-mpu"
+ val path = "data/big.bin"
+ stub {
+ case StubRequest("PUT", p, _, _)
+ if p ==
s"/api/v1/repositories/$repo/branches/main/staging/pmpu/upload-1" =>
+ StubResponse(200, objectStats(path, 3L, "s3://bucket/phys"))
+ }
+
+ val stats = LakeFSStorageClient.completePresignedMultipartUploads(
+ repo,
+ path,
+ "upload-1",
+ // Callers collect part ETags concurrently, so they arrive shuffled; S3
rejects a
+ // completion whose parts are not in ascending part-number order.
+ List(3 -> "etag-3", 1 -> "etag-1", 2 -> "etag-2"),
+ "s3://bucket/phys"
+ )
+ stats.getPath shouldEqual path
+
+ val body = json(onlyRequest.body)
+ val parts = body.get("parts").elements().asScala.toList
+ parts.map(_.get("part_number").asInt()) shouldEqual List(1, 2, 3)
+ // ETags must ride along with their own part number, not just be sorted
independently.
+ parts.map(_.get("etag").asText()) shouldEqual List("etag-1", "etag-2",
"etag-3")
+ body.get("physical_address").asText() shouldEqual "s3://bucket/phys"
+ onlyRequest.query.get("path") shouldBe Some(path)
+ }
+
+ "abortPresignedMultipartUploads" should "delete the upload and name its
physical address" in {
+ val repo = "texera-mpu"
+ val path = "data/big.bin"
+ stub {
+ case StubRequest("DELETE", p, _, _)
+ if p ==
s"/api/v1/repositories/$repo/branches/main/staging/pmpu/upload-1" =>
+ StubResponse(204)
+ }
+
+ LakeFSStorageClient.abortPresignedMultipartUploads(repo, path, "upload-1",
"s3://bucket/phys")
+
+ // lakeFS needs the physical address to clean up the backing object;
omitting it leaks the
+ // multipart parts in the bucket.
+ json(onlyRequest.body).get("physical_address").asText() shouldEqual
"s3://bucket/phys"
+ onlyRequest.query.get("path") shouldBe Some(path)
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Branch and object mutations
+ //
---------------------------------------------------------------------------------------------
+
+ "resetObjectUploadOrDeletion" should "reset exactly one object path on the
main branch" in {
+ val repo = "texera-reset"
+ stub {
+ case StubRequest("PUT", p, _, _) if p ==
s"/api/v1/repositories/$repo/branches/main" =>
+ StubResponse(204)
+ }
+
+ LakeFSStorageClient.resetObjectUploadOrDeletion(repo, "staged/one.bin")
+
+ val body = json(onlyRequest.body)
+ // "object" scopes the reset to the single path. The other ResetCreation
types ("common_prefix"
+ // and "reset") would discard other users' staged work on the same branch.
+ body.get("type").asText() shouldEqual "object"
+ body.get("path").asText() shouldEqual "staged/one.bin"
+ }
+
+ "createCommit" should "commit on the caller-supplied branch" in {
+ val repo = "texera-commit"
+ stub {
+ case StubRequest("POST", p, _, _)
+ if p == s"/api/v1/repositories/$repo/branches/feature-x/commits" =>
+ StubResponse(201, commit("commit-1"))
+ }
+
+ // Unlike the other mutating calls in this class, createCommit is
branch-parameterised; falling
+ // back to the hard-coded main branch would commit to the wrong place.
+ LakeFSStorageClient.createCommit(repo, "feature-x", "a message").getId
shouldEqual "commit-1"
+ json(onlyRequest.body).get("message").asText() shouldEqual "a message"
+ }
+
+ "deleteObject" should "delete the staged path on the main branch" in {
+ val repo = "texera-delete"
+ stub {
+ case StubRequest("DELETE", p, _, _)
+ if p == s"/api/v1/repositories/$repo/branches/main/objects" =>
+ StubResponse(204)
+ }
+
+ LakeFSStorageClient.deleteObject(repo, "staged/one.bin")
+ onlyRequest.query.get("path") shouldBe Some("staged/one.bin")
+ }
+
+ // Left uncovered on purpose:
+ // - `deleteRepo` is a one-line passthrough with no choice of its own to
assert, so a test
+ // could only restate its body;
+ // - `removeFileFromRepo` and `retrieveFileContent` have no caller
anywhere in the repo, and
+ // `withCreateVersion` only has one in `file-service`'s
DatasetResourceSpec — testing them
+ // here would make dead code harder to delete.
+
+ //
---------------------------------------------------------------------------------------------
+ // Read-side getters
+ //
---------------------------------------------------------------------------------------------
+
+ "getFilePresignedUrl" should "ask for a presigned stat and return the
physical address" in {
+ val repo = "texera-presign"
+ val hash = "commit-abc"
+ stub {
+ case StubRequest("GET", p, _, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/stat" =>
+ StubResponse(200, objectStats("data/x.csv", 12L,
"https://signed.example/x.csv?sig=1";))
+ }
+
+ // The presigned URL is the physical address of a *presigned* stat;
without presign=true lakeFS
+ // returns the raw storage address, which the browser cannot fetch.
+ LakeFSStorageClient.getFilePresignedUrl(repo, hash, "data/x.csv")
shouldEqual
+ "https://signed.example/x.csv?sig=1";
+ onlyRequest.query.get("presign") shouldBe Some("true")
+ onlyRequest.query.get("path") shouldBe Some("data/x.csv")
+ }
+
+ "getFileSize" should "return the object size at the given commit" in {
+ val repo = "texera-size-of"
+ val hash = "commit-abc"
+ stub {
+ case StubRequest("GET", p, _, _)
+ if p == s"/api/v1/repositories/$repo/refs/$hash/objects/stat" =>
+ // mtime in the fixture is 1700000000; returning it instead of
size_bytes would be a
+ // plausible slip, and this value is far from it.
+ StubResponse(200, objectStats("data/x.csv", 4242L, "s3://bucket/x"))
+ }
+
+ LakeFSStorageClient.getFileSize(repo, hash, "data/x.csv") shouldEqual 4242L
+ onlyRequest.query.get("path") shouldBe Some("data/x.csv")
+ }
+
+ "getFileFromRepo" should "download the object at the requested version, not
the main branch" in {
+ val repo = "texera-get"
+ val hash = "commit-abc"
+ val content = "hello-lakefs"
+ stub {
+ case StubRequest("GET", p, _, _) if p ==
s"/api/v1/repositories/$repo/refs/$hash/objects" =>
+ StubResponse(200, content)
+ }
+
+ val file = LakeFSStorageClient.getFileFromRepo(repo, hash, "data/x.csv")
+ try {
+ new String(Files.readAllBytes(file.toPath), UTF_8) shouldEqual content
+ // Reading a historical version off the mutable main branch would
silently return the wrong
+ // bytes; the ref segment has to be the commit hash.
+ onlyRequest.path shouldEqual
s"/api/v1/repositories/$repo/refs/$hash/objects"
+ onlyRequest.query.get("path") shouldBe Some("data/x.csv")
+ } finally Files.deleteIfExists(file.toPath)
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // healthCheck
+ //
---------------------------------------------------------------------------------------------
+
+ "healthCheck" should "return after a single probe when the server is
healthy" in {
+ stub {
+ case StubRequest("GET", "/api/v1/healthcheck", _, _) => StubResponse(204)
+ }
+
+ LakeFSStorageClient.healthCheck()
+
+ // The retry wrapper must not re-probe a server that answered; FileService
calls this on
+ // startup, and an unconditional retry would add seconds to every boot.
+ onlyRequest.path shouldEqual "/api/v1/healthcheck"
+ }
}
