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-6987-af1d73ecfc723d6b32265fc76a0382efbea0b0fd in repository https://gitbox.apache.org/repos/asf/texera.git
commit 234270b32857e70f61e0dcc4701c1053bf13095e Author: Xinyuan Lin <[email protected]> AuthorDate: Tue Jul 28 22:49:22 2026 -0700 test(workflow-core): extend util and tuple unit test coverage (#6987) ### What changes were proposed in this PR? Adds 53 tests across four `common/workflow-core` classes (88 -> 141 in these suites). No source changes. - **`StackTraceUtils`** (was **0% covered**, new spec): pins the exact rendered output (header + `Throwable.toString` + frames), the `Caused by` recursion including a 20-deep chain, per-cause frames, and the null-message case. - **`AttributeTypeUtils`**: the unsupported-type `case _` branches of `parseInteger`/`parseLong`/`parseDouble`/`parseBoolean` (asserting both the message and the wrapped `IllegalArgumentException` cause), the previously-unexercised Long/Integer/Timestamp/Boolean input arms, forced-vs-strict divergence (forced `"2147483648"` wraps to `Int.MinValue`; `"1.9"` truncates where strict throws), the `trim.toInt == 1` fallback, schema-inference edges (empty, one-way widening, all-null), and `tupleCasting` null handling. - **`S3StorageClient`**: `uploadPartWithRequest`'s two payload branches (known `contentLength` -> `fromInputStream`, unknown -> `readAllBytes`) with a two-part ordered round trip, the empty-prefix `require` guard, and the abort-on-mid-stream-failure path — asserting the exception propagates, no pending upload id remains (proving `abortMultipartUpload` ran), and no object was published. Follows the suite's existing MinIO-testcontainer approach (no client double exists); no live AWS. - **`LargeBinaryManager`**: `baseUriForExecution` formatting, S3-verified re-seeding across executions, clear-on-empty/null making `create()` fail loudly, and thread-local isolation. Two branches were deliberately left uncovered rather than forced, and are called out in the code review notes: the `DEFAULT_BUCKET.isEmpty` arm (a val captured at object init — unreachable without mutating production config) and a real per-key `DeleteObjects` failure (needs S3 object-lock setup; `throwOnDeleteErrors` is already tested directly). ### Any related issues, documentation, discussions? Closes #6985. ### How was this PR tested? `sbt -java-home <jbr-17> "WorkflowCore/testOnly *S3StorageClientSpec *AttributeTypeUtilsSpec *StackTraceUtilsSpec *LargeBinaryManagerSpec"` -> 141 succeeded, 0 failed (4 suites). `Test/scalafmtCheck` + `Test/scalafix --check` clean. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context]) --- .../amber/core/tuple/AttributeTypeUtilsSpec.scala | 227 ++++++++++++++++ .../texera/amber/util/StackTraceUtilsSpec.scala | 143 +++++++++++ .../service/util/LargeBinaryManagerSpec.scala | 76 ++++++ .../texera/service/util/S3StorageClientSpec.scala | 284 ++++++++++++++++++++- 4 files changed, 727 insertions(+), 3 deletions(-) diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala index b6ec7e5c79..38ba986256 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala @@ -475,4 +475,231 @@ class AttributeTypeUtilsSpec extends AnyFunSuite { } assert(longEx.getMessage == "Failed to parse type java.lang.String to Long: abc") } + + test("parseField returns null for every target type when the field is null") { + // The null short-circuit runs before any per-type parsing. + Seq(INTEGER, LONG, DOUBLE, BOOLEAN, TIMESTAMP, STRING, BINARY, LARGE_BINARY, ANY).foreach { + attributeType => assert(parseField(null, attributeType) == null) + } + } + + test("parseField to INTEGER accepts Long input and trims surrounding whitespace") { + assert(parseField(java.lang.Long.valueOf(42L), INTEGER) == 42) + // A Long beyond Int range is narrowed rather than rejected. + assert(parseField(java.lang.Long.valueOf(3000000000L), INTEGER) == 3000000000L.toInt) + assert(parseField(" 7\t\n", INTEGER) == 7) + assert(parseField(" 1,234 ", INTEGER, force = true) == 1234) + } + + test("parseField to INTEGER rejects types that have no numeric meaning") { + // Timestamp / binary fall into the unsupported branch and surface as AttributeTypeException. + val timestampEx = intercept[AttributeTypeException] { + parseField(new Timestamp(0L), INTEGER) + } + assert(timestampEx.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Integer:")) + assert(timestampEx.getCause.isInstanceOf[IllegalArgumentException]) + assert( + timestampEx.getCause.getMessage == + "Unsupported type for parsing to Integer: java.sql.Timestamp" + ) + + assertThrows[AttributeTypeException](parseField(Array[Byte](1, 2), INTEGER)) + } + + test("parseField to INTEGER overflows out of range values only under forced parsing") { + // Strict parsing rejects a value beyond Int range... + assertThrows[AttributeTypeException](parseField("2147483648", INTEGER)) + // ...while the forced (locale-aware) path parses a Long and narrows it, wrapping around. + assert(parseField("2147483648", INTEGER, force = true) == Integer.MIN_VALUE) + // Forced parsing also accepts decimals and truncates toward zero, unlike strict parsing. + assertThrows[AttributeTypeException](parseField("1.9", INTEGER)) + assert(parseField("1.9", INTEGER, force = true) == 1) + assert(parseField("-1.9", INTEGER, force = true) == -1) + } + + test("parseField to LONG accepts Integer and Timestamp inputs") { + assert(parseField(java.lang.Integer.valueOf(42), LONG) == 42L) + assert(parseField(new Timestamp(1699820130000L), LONG) == 1699820130000L) + assert(parseField(false, LONG) == 0L) + assert(parseField(" -9223372036854775808 ", LONG) == Long.MinValue) + // Forced parsing truncates a decimal instead of failing. + assertThrows[AttributeTypeException](parseField("2.9", LONG)) + assert(parseField("2.9", LONG, force = true) == 2L) + } + + test("parseField to LONG rejects binary input") { + val ex = intercept[AttributeTypeException] { + parseField(Array[Byte](1, 2), LONG) + } + assert(ex.getCause.isInstanceOf[IllegalArgumentException]) + assert(ex.getCause.getMessage.startsWith("Unsupported type for parsing to Long:")) + } + + test("parseField to DOUBLE keeps special float values and rejects timestamps") { + assert(parseField(2.5d, DOUBLE) == 2.5d) + assert(parseField(" -0.75 ", DOUBLE) == -0.75d) + assert(parseField("NaN", DOUBLE).asInstanceOf[java.lang.Double].isNaN) + assert(parseField("Infinity", DOUBLE) == java.lang.Double.POSITIVE_INFINITY) + assert(parseField(false, DOUBLE) == 0.0d) + + val ex = intercept[AttributeTypeException] { + parseField(new Timestamp(0L), DOUBLE) + } + assert(ex.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Double:")) + } + + test("parseField to BOOLEAN maps non-zero numbers to true") { + assert(parseField(2L, BOOLEAN) == true) + assert(parseField(0L, BOOLEAN) == false) + assert(parseField(0.5d, BOOLEAN) == true) + assert(parseField(0.0d, BOOLEAN) == false) + assert(parseField(true, BOOLEAN) == true) + assert(parseField(-1, BOOLEAN) == true) + } + + test("parseField to BOOLEAN treats only the literal 1 as a true numeric string") { + // The fallback is `trim.toInt == 1`, so any other integer string parses to false + // rather than failing. + assert(parseField(" 1 ", BOOLEAN) == true) + assert(parseField("2", BOOLEAN) == false) + assert(parseField("-1", BOOLEAN) == false) + // A non-boolean, non-integer string still fails. + assertThrows[AttributeTypeException](parseField("yes", BOOLEAN)) + } + + test("parseField to BOOLEAN rejects timestamp input") { + val ex = intercept[AttributeTypeException] { + parseField(new Timestamp(0L), BOOLEAN) + } + assert(ex.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Boolean:")) + } + + test("parseField to TIMESTAMP rejects integers and booleans") { + assertThrows[AttributeTypeException](parseField(123, TIMESTAMP)) + assertThrows[AttributeTypeException](parseField(true, TIMESTAMP)) + // Blank strings carry no date at all. + assertThrows[AttributeTypeException](parseField(" ", TIMESTAMP)) + } + + test("parseField to LARGE_BINARY rejects a value that is not an s3 URI") { + // LargeBinary validates its URI scheme, so the failure is an IllegalArgumentException + // raised by the constructor rather than an AttributeTypeException. + val ex = intercept[IllegalArgumentException] { + parseField("/local/path", LARGE_BINARY) + } + assert(ex.getMessage == "LargeBinary URI must start with 's3://', got: /local/path") + assert(!ex.isInstanceOf[AttributeTypeException]) + } + + test("parseField to STRING and BINARY passes values through untouched") { + val binary = Array[Byte](9, 8, 7) + // BINARY is returned by reference, not copied or re-encoded. + assert(parseField(binary, BINARY).asInstanceOf[Array[Byte]] eq binary) + // STRING relies on toString, including for timestamps. + assert(parseField(new Timestamp(0L), STRING) == new Timestamp(0L).toString) + assert(parseField(Seq(1, 2), ANY) == Seq(1, 2)) + } + + test("inferSchemaFromRows returns no types for an empty iterator") { + assert(inferSchemaFromRows(Iterator.empty).isEmpty) + } + + test("inferSchemaFromRows widens a column only in one direction") { + // Once a column has widened to STRING it never narrows back, even if later rows + // would parse as integers on their own. + val rows: Iterator[Array[Any]] = Iterator( + Array[Any]("1", "1"), + Array[Any]("not-a-number-or-date", "2"), + Array[Any]("3", "3") + ) + val attributeTypes = inferSchemaFromRows(rows) + assert(attributeTypes(0) == STRING) + assert(attributeTypes(1) == INTEGER) + } + + test("inferSchemaFromRows keeps INTEGER for a column of only nulls") { + // Null carries no evidence, so the seed type survives. + val rows: Iterator[Array[Any]] = Iterator(Array[Any](null), Array[Any](null)) + assert(inferSchemaFromRows(rows).sameElements(Array(INTEGER))) + } + + test("inferField with no prior type falls back through the type ladder") { + assert(inferField(null) == INTEGER) + assert(inferField("9223372036854775807") == LONG) + assert(inferField("not parseable at all") == STRING) + } + + test("inferField never narrows a STRING column back to a parseable type") { + assert(inferField(STRING, "1") == STRING) + assert(inferField(STRING, null) == STRING) + } + + test("compare rejects the ANY type") { + val ex = intercept[UnsupportedOperationException] { + compare(1, 2, ANY) + } + // ANY renders as the empty string (it is hidden from the JSON schema), so the message + // names no type. + assert(ex.getMessage == "Unsupported attribute type for compare: ") + } + + test("compare orders binary arrays of differing lengths by prefix") { + // A prefix sorts before the longer array that extends it. + assert(compare(Array[Byte](1, 2), Array[Byte](1, 2, 0), BINARY) < 0) + assert(compare(Array.emptyByteArray, Array[Byte](0), BINARY) < 0) + } + + test("compare places NaN above every other double") { + assert(compare(java.lang.Double.NaN, java.lang.Double.MAX_VALUE, DOUBLE) > 0) + assert(compare(java.lang.Double.NEGATIVE_INFINITY, 0.0d, DOUBLE) < 0) + } + + test("add with two null operands fails for types that have no zero value") { + // The null/null branch delegates to zeroValue, which rejects unsupported types. + val ex = intercept[UnsupportedOperationException] { + add(null, null, STRING) + } + assert(ex.getMessage == "Unsupported attribute type for zero value: string") + } + + test("add returns the non-null operand unchanged for otherwise unsupported types") { + // The identity branches run before the per-type dispatch, so no exception is raised. + assert(add("kept", null, STRING) == "kept") + assert(add(null, "kept", STRING) == "kept") + } + + test("SchemaCasting leaves the schema untouched when no attribute matches") { + val schema = Schema(List(new Attribute("a", STRING), new Attribute("b", STRING))) + assert(SchemaCasting(schema, "missing", INTEGER) == schema) + } + + test("SchemaCasting retains the original type for LARGE_BINARY targets") { + // LARGE_BINARY is outside the supported cast list, so the attribute keeps its type. + val schema = Schema(List(new Attribute("a", STRING))) + assert(SchemaCasting(schema, "a", LARGE_BINARY) == schema) + } + + test("SchemaCasting supports every castable target type") { + val schema = Schema(List(new Attribute("a", STRING))) + Seq(STRING, INTEGER, DOUBLE, LONG, BOOLEAN, TIMESTAMP, BINARY).foreach { target => + assert(SchemaCasting(schema, "a", target).getAttribute("a").getType == target) + } + } + + test("tupleCasting preserves nulls and leaves unlisted columns alone") { + val schema = Schema(List(new Attribute("a", STRING), new Attribute("b", STRING))) + val tuple = Tuple.builder(schema).addSequentially(Array[Any](null, "42")).build() + val result = tupleCasting(tuple, Map("a" -> INTEGER)) + assert(result.getFields.sameElements(Array[Any](null, "42"))) + } + + test("tupleCasting propagates a parse failure for an uncastable value") { + val schema = Schema(List(new Attribute("a", STRING))) + val tuple = Tuple.builder(schema).addSequentially(Array[Any]("abc")).build() + assertThrows[AttributeTypeException](tupleCasting(tuple, Map("a" -> INTEGER))) + } + + test("parseFields returns an empty array for empty input") { + assert(parseFields(Array.empty[Any], Array.empty[AttributeType]).isEmpty) + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala new file mode 100644 index 0000000000..afb23300d3 --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala @@ -0,0 +1,143 @@ +/* + * 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.amber.util + +import org.scalatest.funsuite.AnyFunSuite + +class StackTraceUtilsSpec extends AnyFunSuite { + + private val TopLevelHeader = "Stack trace for developers: \n\n" + private val CauseHeader = "\n\nCaused by:\n" + + /** Builds a throwable with no frames so the rendering can be asserted exactly. */ + private def frameless(err: Throwable): Throwable = { + err.setStackTrace(Array.empty) + err + } + + private def occurrencesOf(haystack: String, needle: String): Int = + haystack.sliding(needle.length).count(_ == needle) + + test("getStackTraceWithAllCauses renders a causeless throwable exactly") { + val err = frameless(new RuntimeException("boom")) + + // header + toString + "\n" + (no frames) is the whole output. + assert( + StackTraceUtils.getStackTraceWithAllCauses(err) == + s"${TopLevelHeader}java.lang.RuntimeException: boom\n" + ) + } + + test("getStackTraceWithAllCauses includes every real stack frame, newline-separated") { + val err = new RuntimeException("with frames") + val rendered = StackTraceUtils.getStackTraceWithAllCauses(err) + + assert(rendered.startsWith(TopLevelHeader)) + assert(rendered.contains("java.lang.RuntimeException: with frames")) + // Frames are rendered verbatim, in order, after the throwable's own toString. + val frameBlock = rendered.stripPrefix(s"${TopLevelHeader}${err.toString}\n") + assert(frameBlock == err.getStackTrace.mkString("\n")) + assert(err.getStackTrace.length > 1) + assert(frameBlock.contains(this.getClass.getName)) + } + + test("getStackTraceWithAllCauses omits the Caused-by section when there is no cause") { + val rendered = StackTraceUtils.getStackTraceWithAllCauses(new RuntimeException("lonely")) + assert(!rendered.contains("Caused by:")) + } + + test("getStackTraceWithAllCauses with topLevel=false swaps in the Caused-by header") { + val err = frameless(new IllegalStateException("nested")) + + assert( + StackTraceUtils.getStackTraceWithAllCauses(err, topLevel = false) == + s"${CauseHeader}java.lang.IllegalStateException: nested\n" + ) + assert( + !StackTraceUtils + .getStackTraceWithAllCauses(err, topLevel = false) + .contains("Stack trace for developers") + ) + } + + test("getStackTraceWithAllCauses appends a single cause below the top-level trace") { + val cause = frameless(new IllegalArgumentException("inner")) + val err = frameless(new RuntimeException("outer", cause)) + + assert( + StackTraceUtils.getStackTraceWithAllCauses(err) == + s"${TopLevelHeader}java.lang.RuntimeException: outer\n" + + s"${CauseHeader}java.lang.IllegalArgumentException: inner\n" + ) + } + + test("getStackTraceWithAllCauses walks the whole chain, outermost first") { + val root = frameless(new java.io.IOException("root")) + val middle = frameless(new IllegalStateException("middle", root)) + val top = frameless(new RuntimeException("top", middle)) + + val rendered = StackTraceUtils.getStackTraceWithAllCauses(top) + + // One developer header for the whole chain, one Caused-by per cause. + assert(occurrencesOf(rendered, "Stack trace for developers") == 1) + assert(occurrencesOf(rendered, "Caused by:") == 2) + // Ordering is outermost -> innermost. + assert(rendered.indexOf("top") < rendered.indexOf("middle")) + assert(rendered.indexOf("middle") < rendered.indexOf("root")) + assert(rendered.endsWith("java.io.IOException: root\n")) + } + + test("getStackTraceWithAllCauses renders each cause's own frames, not just the top's") { + val cause = new IllegalStateException("inner") + val err = new RuntimeException("outer", cause) + + val rendered = StackTraceUtils.getStackTraceWithAllCauses(err) + + assert(rendered.contains(err.getStackTrace.head.toString)) + assert(rendered.contains(cause.getStackTrace.head.toString)) + // The cause's frames belong to the section that follows the Caused-by marker. + val causeSection = rendered.substring(rendered.indexOf(CauseHeader)) + assert(causeSection.contains(cause.getStackTrace.head.toString)) + } + + test("getStackTraceWithAllCauses handles a throwable without a message") { + val err = frameless(new RuntimeException()) + + // Throwable.toString drops the ": <msg>" suffix when the message is null. + assert( + StackTraceUtils.getStackTraceWithAllCauses(err) == + s"${TopLevelHeader}java.lang.RuntimeException\n" + ) + } + + test("getStackTraceWithAllCauses renders a deep cause chain without truncating it") { + // The recursion is unbounded by design; a long chain must be rendered in full. + val root = frameless(new RuntimeException("root")) + val chain = (1 to 20).foldLeft[Throwable](root) { (cause, i) => + frameless(new RuntimeException(s"level-$i", cause)) + } + + val rendered = StackTraceUtils.getStackTraceWithAllCauses(chain) + + assert(occurrencesOf(rendered, "Caused by:") == 20) + assert(rendered.startsWith(TopLevelHeader)) + assert(rendered.endsWith("java.lang.RuntimeException: root\n")) + } +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala index 0f2cce8b30..eeb1bce7cd 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala @@ -523,4 +523,80 @@ class LargeBinaryManagerSpec extends AnyFunSuite with S3StorageTestBase with Bef setExecutionContext(TestExecutionId) } } + + // ======================================== + // Base-URI Tests + // ======================================== + + test("baseUriForExecution scopes binaries to the configured bucket and the execution") { + assert( + LargeBinaryManager.baseUriForExecution(4242L) == + s"s3://${LargeBinaryManager.DEFAULT_BUCKET}/objects/4242/" + ) + // The trailing slash is what keeps execution 4242 and 42421 in separate prefixes. + assert(LargeBinaryManager.baseUriForExecution(4242L).endsWith("/")) + assert(LargeBinaryManager.DEFAULT_BUCKET == "texera-large-binaries") + } + + test("re-seeding the base URI routes later binaries to the new execution's prefix") { + setExecutionContext(3001L) + val first = createLargeBinary("written under 3001") + setExecutionContext(3002L) + val second = createLargeBinary("written under 3002") + + try { + assert(first.getObjectKey.startsWith("objects/3001/")) + assert(second.getObjectKey.startsWith("objects/3002/")) + // Both objects really landed under their own execution prefix in S3. + assert(S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, "objects/3001")) + assert(S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, "objects/3002")) + } finally { + LargeBinaryManager.deleteByExecution(3001L) + LargeBinaryManager.deleteByExecution(3002L) + setExecutionContext(TestExecutionId) + } + } + + test("an empty base URI clears the thread's context so create fails loudly") { + // An unconfigured bucket yields an empty base URI; create() must not silently produce + // a bucket-less key. + try { + LargeBinaryManager.setCurrentBaseUri("") + val thrown = intercept[IllegalStateException](LargeBinaryManager.create()) + assert(thrown.getMessage.contains("requires a base URI")) + } finally { + setExecutionContext(TestExecutionId) + } + } + + test("a null base URI clears the thread's context so create fails loudly") { + try { + LargeBinaryManager.setCurrentBaseUri(null) + assertThrows[IllegalStateException](LargeBinaryManager.create()) + // new LargeBinary() delegates to create(), so it fails the same way. + assertThrows[IllegalStateException](new LargeBinary()) + } finally { + setExecutionContext(TestExecutionId) + } + } + + test("the base URI is thread-local and does not leak into other threads") { + // create() runs on the DP thread; a sibling thread must not inherit this thread's context. + setExecutionContext(TestExecutionId) + @volatile var caught: Option[Throwable] = None + val other = new Thread(() => { + try LargeBinaryManager.create() + catch { case e: Throwable => caught = Some(e) } + }) + other.start() + other.join() + + assert(caught.exists(_.isInstanceOf[IllegalStateException])) + // This thread's context is unaffected. + assert( + LargeBinaryManager + .create() + .startsWith(LargeBinaryManager.baseUriForExecution(TestExecutionId)) + ) + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala index c98f3c589c..bcd50dc23d 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala @@ -19,14 +19,19 @@ package org.apache.texera.service.util +import org.apache.texera.common.config.StorageConfig import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} import org.scalatest.funsuite.AnyFunSuite -import software.amazon.awssdk.services.s3.model.S3Error +import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider} +import software.amazon.awssdk.regions.Region +import software.amazon.awssdk.services.s3.model._ +import software.amazon.awssdk.services.s3.{S3Client, S3Configuration} -import java.io.ByteArrayInputStream +import java.io.{ByteArrayInputStream, IOException, InputStream} import java.util.concurrent.Executors import scala.concurrent.duration._ import scala.concurrent.{Await, ExecutionContext, Future} +import scala.jdk.CollectionConverters._ import scala.util.Random class S3StorageClientSpec @@ -45,13 +50,64 @@ class S3StorageClientSpec override def afterAll(): Unit = { // Best-effort cleanup of the prefixes these tests use (deleteDirectory rejects an empty prefix). try { - Seq("test", "delete-dir").foreach(S3StorageClient.deleteDirectory(testBucketName, _)) + Seq("test", "delete-dir", "multipart") + .foreach(S3StorageClient.deleteDirectory(testBucketName, _)) } catch { case _: Exception => // Ignore cleanup errors } + try rawClient.close() + catch { case _: Exception => } super.afterAll() } + /** + * A second S3 client pointed at the same MinIO container. S3StorageClient exposes only + * `uploadPartWithRequest` from the multipart API, so the surrounding create/complete/list + * calls are issued directly instead of being mocked away. + */ + private lazy val rawClient: S3Client = { + val credentials = AwsBasicCredentials.create(StorageConfig.s3Username, StorageConfig.s3Password) + S3Client + .builder() + .credentialsProvider(StaticCredentialsProvider.create(credentials)) + .region(Region.of(StorageConfig.s3Region)) + .endpointOverride(java.net.URI.create(StorageConfig.s3Endpoint)) + .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build()) + .build() + } + + private def startMultipartUpload(objectKey: String): String = + rawClient + .createMultipartUpload( + CreateMultipartUploadRequest.builder().bucket(testBucketName).key(objectKey).build() + ) + .uploadId() + + private def completeMultipartUpload( + objectKey: String, + uploadId: String, + parts: Seq[CompletedPart] + ): Unit = + rawClient.completeMultipartUpload( + CompleteMultipartUploadRequest + .builder() + .bucket(testBucketName) + .key(objectKey) + .uploadId(uploadId) + .multipartUpload(CompletedMultipartUpload.builder().parts(parts.asJava).build()) + .build() + ) + + /** Upload ids still pending (i.e. neither completed nor aborted) for the given key. */ + private def pendingUploadIds(objectKey: String): Seq[String] = + rawClient + .listMultipartUploads(ListMultipartUploadsRequest.builder().bucket(testBucketName).build()) + .uploads() + .asScala + .filter(_.key() == objectKey) + .map(_.uploadId()) + .toSeq + // Helper methods private def createInputStream(data: String): ByteArrayInputStream = { new ByteArrayInputStream(data.getBytes) @@ -438,4 +494,226 @@ class S3StorageClientSpec assert(!thrown.getMessage.contains(f"delete-dir/locked-$cap%02d.txt")) // capped key is not assert(thrown.getMessage.contains(s"and ${errorCount - cap} more")) } + + test("deleteDirectory should reject an empty prefix instead of wiping the bucket") { + val guard = "delete-dir/guarded-object.txt" + S3StorageClient.uploadObject(testBucketName, guard, createInputStream("keep me")) + + val thrown = intercept[IllegalArgumentException] { + S3StorageClient.deleteDirectory(testBucketName, "") + } + assert(thrown.getMessage.contains("directoryPrefix must not be empty")) + + // The bucket is untouched. + val survivor = S3StorageClient.downloadObject(testBucketName, guard) + assert(new String(readInputStream(survivor)) == "keep me") + survivor.close() + S3StorageClient.deleteObject(testBucketName, guard) + } + + // ======================================== + // directoryExists Tests + // ======================================== + + test("directoryExists should treat a prefix with and without a trailing slash the same") { + val prefix = "test/exists-dir" + S3StorageClient.uploadObject( + testBucketName, + s"$prefix/object.txt", + createInputStream("data") + ) + + assert(S3StorageClient.directoryExists(testBucketName, prefix)) + assert(S3StorageClient.directoryExists(testBucketName, s"$prefix/")) + + S3StorageClient.deleteObject(testBucketName, s"$prefix/object.txt") + } + + test("directoryExists should be false for a prefix that only matches an object name") { + // "test/exists-file" names an object, not a directory, so no key starts with + // "test/exists-file/". + val objectKey = "test/exists-file" + S3StorageClient.uploadObject(testBucketName, objectKey, createInputStream("data")) + + assert(!S3StorageClient.directoryExists(testBucketName, objectKey)) + + S3StorageClient.deleteObject(testBucketName, objectKey) + } + + test("directoryExists should be false for an unused prefix") { + assert(!S3StorageClient.directoryExists(testBucketName, "test/never-written")) + } + + // ======================================== + // createBucketIfNotExist Tests + // ======================================== + + test("createBucketIfNotExist should be idempotent for an existing bucket") { + // beforeAll already created it; a second call takes the headBucket-succeeds path. + S3StorageClient.createBucketIfNotExist(testBucketName) + S3StorageClient.createBucketIfNotExist(testBucketName) + + // The bucket is still usable afterwards. + val objectKey = "test/after-recreate.txt" + S3StorageClient.uploadObject(testBucketName, objectKey, createInputStream("still here")) + val stream = S3StorageClient.downloadObject(testBucketName, objectKey) + assert(new String(readInputStream(stream)) == "still here") + stream.close() + S3StorageClient.deleteObject(testBucketName, objectKey) + } + + // ======================================== + // uploadPartWithRequest Tests + // ======================================== + + test("uploadPartWithRequest should stream a part when the content length is known") { + val objectKey = "multipart/known-length.bin" + val partData = "part payload with a known length".getBytes + val uploadId = startMultipartUpload(objectKey) + + val response = S3StorageClient.uploadPartWithRequest( + testBucketName, + objectKey, + uploadId, + partNumber = 1, + createInputStream(partData), + contentLength = Some(partData.length.toLong) + ) + + assert(response.eTag() != null && response.eTag().nonEmpty) + completeMultipartUpload( + objectKey, + uploadId, + Seq(CompletedPart.builder().partNumber(1).eTag(response.eTag()).build()) + ) + + val stream = S3StorageClient.downloadObject(testBucketName, objectKey) + assert(readInputStream(stream).sameElements(partData)) + stream.close() + S3StorageClient.deleteObject(testBucketName, objectKey) + } + + test("uploadPartWithRequest should buffer the whole stream when no content length is given") { + val objectKey = "multipart/unknown-length.bin" + val partData = "part payload with no declared length".getBytes + val uploadId = startMultipartUpload(objectKey) + + val response = S3StorageClient.uploadPartWithRequest( + testBucketName, + objectKey, + uploadId, + partNumber = 1, + createInputStream(partData), + contentLength = None + ) + + completeMultipartUpload( + objectKey, + uploadId, + Seq(CompletedPart.builder().partNumber(1).eTag(response.eTag()).build()) + ) + + val stream = S3StorageClient.downloadObject(testBucketName, objectKey) + assert(readInputStream(stream).sameElements(partData)) + stream.close() + S3StorageClient.deleteObject(testBucketName, objectKey) + } + + test("uploadPartWithRequest should assemble parts in part-number order") { + // Two parts: every part but the last must be at least 5 MiB, so the head is sized exactly + // at the multipart minimum and the tail is short. + val objectKey = "multipart/ordered-parts.bin" + val head = Array.fill[Byte](S3StorageClient.MINIMUM_NUM_OF_MULTIPART_S3_PART.toInt)(1.toByte) + val tail = Array.fill[Byte](1024)(2.toByte) + val uploadId = startMultipartUpload(objectKey) + + val firstETag = S3StorageClient + .uploadPartWithRequest( + testBucketName, + objectKey, + uploadId, + partNumber = 1, + createInputStream(head), + contentLength = Some(head.length.toLong) + ) + .eTag() + val secondETag = S3StorageClient + .uploadPartWithRequest( + testBucketName, + objectKey, + uploadId, + partNumber = 2, + createInputStream(tail), + contentLength = None + ) + .eTag() + + completeMultipartUpload( + objectKey, + uploadId, + Seq( + CompletedPart.builder().partNumber(1).eTag(firstETag).build(), + CompletedPart.builder().partNumber(2).eTag(secondETag).build() + ) + ) + + val stream = S3StorageClient.downloadObject(testBucketName, objectKey) + val downloaded = readInputStream(stream) + stream.close() + + assert(downloaded.length == head.length + tail.length) + assert(downloaded.take(head.length).sameElements(head)) + assert(downloaded.drop(head.length).sameElements(tail)) + + S3StorageClient.deleteObject(testBucketName, objectKey) + } + + test("uploadPartWithRequest should fail for an unknown upload id") { + val objectKey = "multipart/no-such-upload.bin" + val thrown = intercept[S3Exception] { + S3StorageClient.uploadPartWithRequest( + testBucketName, + objectKey, + uploadId = "not-a-real-upload-id", + partNumber = 1, + createInputStream("payload"), + contentLength = None + ) + } + assert(thrown.statusCode() == 404) + } + + // ======================================== + // uploadObject failure-path Tests + // ======================================== + + test("uploadObject should abort the multipart upload when the source stream fails") { + // Fails only after the first full part has been read, so a multipart upload is already + // in flight when the error surfaces. + val objectKey = "multipart/failing-source.bin" + val firstPart = + Array.fill[Byte](S3StorageClient.MINIMUM_NUM_OF_MULTIPART_S3_PART.toInt)(7.toByte) + + val failingStream = new InputStream { + private val delegate = new ByteArrayInputStream(firstPart) + override def read(): Int = throw new IOException("stream broke") + override def read(b: Array[Byte], off: Int, len: Int): Int = { + val n = delegate.read(b, off, len) + if (n <= 0) throw new IOException("stream broke") else n + } + } + + val thrown = intercept[IOException] { + S3StorageClient.uploadObject(testBucketName, objectKey, failingStream) + } + assert(thrown.getMessage == "stream broke") + + // The in-flight upload was aborted, so no orphaned upload is left behind... + assert(pendingUploadIds(objectKey).isEmpty) + // ...and no object was published under the key. + val missing = intercept[S3Exception] { + S3StorageClient.downloadObject(testBucketName, objectKey) + } + assert(missing.statusCode() == 404) + } }
