chore: cleanup presign tests (#546)

Author: aajtodd

services/build.gradle.kts

@@ -57,6 +57,9 @@ subprojects {
         testLogging {
             events("passed", "skipped", "failed")
             showStandardStreams = true
+            showStackTraces = true
+            showExceptions = true
+            exceptionFormat = org.gradle.api.tasks.testing.logging.TestExceptionFormat.FULL
         }
     }
 
@@ -131,6 +134,9 @@ subprojects {
                     testLogging {
                         events("passed", "skipped", "failed")
                         showStandardStreams = true
+                        showStackTraces = true
+                        showExceptions = true
+                        exceptionFormat = org.gradle.api.tasks.testing.logging.TestExceptionFormat.FULL
                     }
                 }
             }

services/s3/e2eTest/S3PresignerTest.kt

@@ -5,14 +5,11 @@ import aws.sdk.kotlin.services.s3.S3Client
 import aws.sdk.kotlin.services.s3.model.GetObjectRequest
 import aws.sdk.kotlin.services.s3.model.PutObjectRequest
 import aws.sdk.kotlin.services.s3.presigners.presign
-import aws.smithy.kotlin.runtime.content.ByteStream
 import aws.smithy.kotlin.runtime.content.decodeToString
 import aws.smithy.kotlin.runtime.http.response.complete
 import aws.smithy.kotlin.runtime.http.sdkHttpClient
 import aws.smithy.kotlin.runtime.http.toByteStream
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.*
 import org.junit.jupiter.api.AfterAll
 import org.junit.jupiter.api.BeforeAll
 import org.junit.jupiter.api.TestInstance
@@ -27,8 +24,13 @@ class S3PresignerTest {
         const val DEFAULT_REGION = "us-east-2"
     }
 
+    private val engine = CrtHttpEngine {
+        maxConnections = 64u
+    }
+
     private val client = S3Client {
         region = DEFAULT_REGION
+        httpClientEngine = engine
     }
 
     private lateinit var testBucket: String
@@ -39,58 +41,36 @@ class S3PresignerTest {
     }
 
     @AfterAll
-    private fun cleanup() = runBlocking {
+    private fun cleanup(): Unit = runBlocking {
         S3TestUtils.deleteBucketAndAllContents(client, testBucket)
+        client.close()
+        engine.close()
     }
 
     @Test
-    fun testPutObjectPresigner() = runTest {
+    fun testRoundTripBlns() = runBlocking {
         val contents = "presign-test"
+        val dispatcher = Dispatchers.IO.limitedParallelism(64)
+        val httpClient = sdkHttpClient(engine)
 
-        // FIXME - run these concurrently, test takes forever
         testKeyNames.forEach { keyName ->
-            val presignedRequest = PutObjectRequest {
-                bucket = testBucket
-                key = keyName
-            }.presign(client.config, 60.seconds)
-
-            S3TestUtils.responseCodeFromPut(presignedRequest, contents)
-
-            val req = GetObjectRequest {
-                bucket = testBucket
-                key = keyName
-            }
-            val roundTrippedContents = client.getObject(req) { it.body?.decodeToString() }
-
-            assertEquals(contents, roundTrippedContents)
-        }
-    }
+            launch(dispatcher) {
+                val presignedPutRequest = PutObjectRequest {
+                    bucket = testBucket
+                    key = keyName
+                }.presign(client.config, 60.seconds)
 
-    @Test
-    fun testGetObjectPresigner() = runTest {
-        val contents = "presign-test"
-
-        // FIXME - run these concurrently, test takes forever
-        testKeyNames.reversed().forEach { keyName ->
-            client.putObject {
-                bucket = testBucket
-                key = keyName
-                body = ByteStream.fromString(contents)
-            }
+                S3TestUtils.responseCodeFromPut(presignedPutRequest, contents)
 
-            val presignedRequest = GetObjectRequest {
-                bucket = testBucket
-                key = keyName
-            }.presign(client.config, 60.seconds)
+                val presignedGetRequest = GetObjectRequest {
+                    bucket = testBucket
+                    key = keyName
+                }.presign(client.config, 60.seconds)
 
-            CrtHttpEngine().use { engine ->
-                val httpClient = sdkHttpClient(engine)
-
-                val call = httpClient.call(presignedRequest)
+                val call = httpClient.call(presignedGetRequest)
+                val body = call.response.body.toByteStream()?.decodeToString()
                 call.complete()
-
                 assertEquals(200, call.response.status.value)
-                val body = call.response.body.toByteStream()?.decodeToString()
                 assertEquals(contents, body)
             }
         }
@@ -248,7 +228,7 @@ val testKeyNames: List<String>
             # Unicode additional control characters: all of the characters with
             # general category Cf (in Unicode 8.0.0).
             # The next line may appear to be blank or mojibake in some viewers.
-            ­؀؁؂؃؄؅؜۝܏᠎​‌‍‎‏‪‫‬‭‮⁠⁡⁢⁣⁤⁦⁧⁨⁩𑂽𛲠𛲡𛲢𛲣𝅳𝅴𝅵𝅶𝅷𝅸𝅹𝅺󠀁󠀠󠀡󠀢󠀣󠀤󠀥󠀦󠀧󠀨󠀩󠀪󠀫󠀬󠀭󠀮󠀯󠀰󠀱󠀲󠀳󠀴󠀵󠀶󠀷󠀸󠀹󠀺󠀻󠀼󠀽󠀾󠀿󠁀󠁁󠁂󠁃󠁄󠁅󠁆󠁇󠁈󠁉󠁊󠁋󠁌󠁍󠁎󠁏󠁐󠁑󠁒󠁓󠁔󠁕󠁖󠁗󠁘󠁙󠁚󠁛󠁜󠁝󠁞󠁟󠁠󠁡󠁢󠁣󠁤󠁥󠁦󠁧󠁨󠁩󠁪󠁫󠁬󠁭󠁮󠁯󠁰󠁱󠁲󠁳󠁴󠁵󠁶󠁷󠁸󠁹󠁺󠁻󠁼󠁽󠁾󠁿
+            ­؀؁؂؃؄؅؜۝܏᠎​‌‍‎‏‪‫‬‭‮⁠⁡⁢⁣⁤⁦⁧⁨⁩𑂽𛲠𛲡𛲢𛲣𝅳𝅴𝅵𝅶𝅷𝅸𝅹𝅺󠀁󠀠󠀡󠀢󠀣󠀤󠀥󠀦󠀧󠀨󠀩󠀪󠀫󠀬󠀭󠀮󠀯󠀰󠀱󠀲󠀳󠀴󠀵󠀶󠀷󠀸󠀹󠀺󠀻󠀼󠀽󠀾󠀿󠁀󠁁󠁂󠁃󠁄󠁅󠁆󠁇󠁈󠁉󠁊󠁋󠁌󠁍󠁎󠁏󠁐󠁑󠁒󠁓󠁔󠁕󠁖󠁗󠁘󠁙󠁚󠁛󠁜󠁝󠁞󠁟󠁠󠁡󠁢󠁣󠁤󠁥󠁦󠁧󠁨󠁩󠁪󠁫󠁬󠁭󠁮󠁯󠁰󠁱󠁲󠁳󠁴󠁵󠁶󠁷󠁸󠁹󠁺󠁻󠁼󠁽󠁾󠁿
     
             # "Byte order marks", U+FEFF and U+FFFE, each on its own line.
             # The next two lines may appear to be blank or mojibake in some viewers.

services/s3/e2eTest/S3TestUtils.kt

@@ -9,10 +9,11 @@ import aws.sdk.kotlin.services.s3.model.BucketLocationConstraint
 import aws.sdk.kotlin.services.s3.model.ExpirationStatus
 import aws.sdk.kotlin.services.s3.model.LifecycleRule
 import aws.sdk.kotlin.services.s3.model.LifecycleRuleFilter
-import aws.sdk.kotlin.services.s3.model.NotFound
+import aws.sdk.kotlin.services.s3.paginators.listObjectsV2Paginated
+import aws.sdk.kotlin.services.s3.waiters.waitUntilBucketExists
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
 import java.io.OutputStreamWriter
 import java.net.URL
 import java.util.*
@@ -40,15 +41,7 @@ object S3TestUtils {
                 }
             }
 
-            do {
-                val bucketExists = try {
-                    client.headBucket { bucket = testBucket }
-                    true
-                } catch (ex: NotFound) {
-                    delay(300)
-                    false
-                }
-            } while (!bucketExists)
+            client.waitUntilBucketExists { bucket = testBucket }
         }
 
         client.putBucketLifecycleConfiguration {
@@ -68,30 +61,31 @@ object S3TestUtils {
         testBucket
     }
 
-    suspend fun deleteBucketAndAllContents(client: S3Client, bucketName: String) {
+    @OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class)
+    suspend fun deleteBucketAndAllContents(client: S3Client, bucketName: String): Unit = coroutineScope {
+        val scope = this
+
         try {
             println("Deleting S3 bucket: $bucketName")
-
-            var resp = client.listObjectsV2 { bucket = bucketName }
-
-            do {
-                val objects = resp.contents
-                val truncated = resp.isTruncated
-
-                objects?.forEach {
-                    client.deleteObject {
-                        bucket = bucketName
-                        key = it.key
+            val dispatcher = Dispatchers.Default.limitedParallelism(64)
+            val jobs = mutableListOf<Job>()
+
+            // FIXME - this should use the batch `DeleteObjects` request and delete by page rather than individual key
+            //         However the current XML serializer chokes on the BLNS keys used by the presign tests. Update after
+            //         new XML implementation is available
+            client.listObjectsV2Paginated { bucket = bucketName }
+                .flatMapConcat { it.contents?.asFlow() ?: flowOf() }
+                .collect { obj ->
+                    val job = scope.launch(dispatcher) {
+                        client.deleteObject {
+                            bucket = bucketName
+                            key = obj.key
+                        }
                     }
+                    jobs.add(job)
                 }
 
-                if (truncated) {
-                    resp = client.listObjectsV2 {
-                        bucket = bucketName
-                        continuationToken = resp.continuationToken
-                    }
-                }
-            } while (truncated)
+            jobs.joinAll()
 
             client.deleteBucket { bucket = bucketName }
         } catch (ex: Exception) {